/*
 * thr_info.c
 *
 * Copyright (C) 2008-2022 Aerospike, Inc.
 *
 * Portions may be licensed to Aerospike, Inc. under one or more contributor
 * license agreements.
 *
 * This program is free software: you can redistribute it and/or modify it under
 * the terms of the GNU Affero General Public License as published by the Free
 * Software Foundation, either version 3 of the License, or (at your option) any
 * later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
 * details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see http://www.gnu.org/licenses/
 */

//==========================================================
// Includes.
//

#include "base/thr_info.h"

#include <errno.h>
#include <fcntl.h>
#include <getopt.h>
#include <stdarg.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>
#include <ctype.h>
#include <limits.h>
#include <malloc.h>
#include <mcheck.h>
#include <sys/ioctl.h>
#include <sys/resource.h>
#include <time.h>
#include <unistd.h>

#include "aerospike/as_atomic.h"
#include "citrusleaf/alloc.h"
#include "citrusleaf/cf_queue.h"

#include "cf_mutex.h"
#include "cf_str.h"
#include "cf_thread.h"
#include "dns.h"
#include "dynbuf.h"
#include "fetch.h"
#include "hardware.h"
#include "hist.h"
#include "log.h"
#include "msgpack_in.h"
#include "os.h"
#include "shash.h"
#include "socket.h"
#include "vault.h"
#include "vector.h"
#include "xmem.h"

#include "base/batch.h"
#include "base/cfg.h"
#include "base/cfg_info.h"
#include "base/datamodel.h"
#include "base/features.h"
#include "base/health.h"
#include "base/index.h"
#include "base/mrt_monitor.h"
#include "base/nsup.h"
#include "base/proto.h"
#include "base/security.h"
#include "base/service.h"
#include "base/set_index.h"
#include "base/smd.h"
#include "base/stats.h"
#include "base/thr_info_port.h"
#include "base/thr_tsvc.h"
#include "base/transaction.h"
#include "base/truncate.h"
#include "base/udf_cask.h"
#include "base/xdr.h"
#include "fabric/exchange.h"
#include "fabric/fabric.h"
#include "fabric/hb.h"
#include "fabric/hlc.h"
#include "fabric/migrate.h"
#include "fabric/partition.h"
#include "fabric/partition_balance.h"
#include "fabric/roster.h"
#include "fabric/service_list.h"
#include "fabric/skew_monitor.h"
#include "query/query_manager.h"
#include "sindex/sindex.h"
#include "storage/storage.h"
#include "transaction/proxy.h"
#include "transaction/rw_request_hash.h"


//==========================================================
// Typedefs & constants.
//

#define EOL '\n' // incoming commands are separated by EOL
#define SEP '\t' // in the result, sepeates the echoed command from the result
#define ARG_SEP '/' // separates fixed arguments
#define ARG_KWARG_SEP ':' // separates fixed arguments from key/value arguments
#define KWARG_SEP ';' // separates key/value arguments
#define KV_SEP '=' // separates keys from values

#define CMD_NAME_MAX_LEN 32

typedef void (*as_info_cmd_fn)(const char* name, const char* params, cf_dyn_buf* db);

typedef struct as_info_cmd_s {
	const char* name;
	const as_info_cmd_fn fn;
	bool in_summary;
	bool client_only;
	bool ee_only;
	as_sec_perm perm; // required security permission
} as_info_cmd;

typedef struct rack_node_s {
	uint32_t rack_id;
	cf_node node;
} rack_node;

typedef struct find_sindex_key_udata_s {
	const char* ns_name;
	const char* index_name;
	const char* smd_key;
	char* found_key; // only when unique
	uint32_t n_name_matches;
	uint32_t n_indexes;
	bool has_smd_key;
} find_sindex_key_udata;


//==========================================================
// Globals.
//

as_stats g_stats = { 0 }; // separate .c file not worth it
cf_dyn_buf g_bad_practices = { 0 };
uint64_t g_start_sec; // start time of the server
static cf_queue g_info_work_q;
static cf_shash* g_specs;

// Defined in "make_in/version.c" (auto-generated by the build system.)
extern const char aerospike_build_id[];
extern const char aerospike_build_time[];
extern const char aerospike_build_type[];
extern const char aerospike_build_os[];
extern const char aerospike_build_arch[];
extern const char aerospike_build_sha[];
extern const char aerospike_build_ee_sha[];
extern const char aerospike_build_features[];

// TODO: This should move elsewhere.
static uint32_t g_process_cpu_pct = 0;

// TODO: This should move elsewhere.
static uint32_t g_user_cpu_pct = 0;
static uint32_t g_kernel_cpu_pct = 0;


//==========================================================
// Forward declarations.
//

// Info execution.
static void* run_info(void* arg);
static bool authenticate(const as_file_handle* fd_h, cf_dyn_buf* db);
static void append_security_error(cf_dyn_buf* db, uint32_t result, as_sec_perm perm);
static void info_summary(cf_dyn_buf* db);
static void handle_cmds(char* buf, size_t buf_sz, const as_file_handle* fd_h, cf_dyn_buf* db);
static const as_info_cmd* parse_cmd(char** cmd_str_p, cf_dyn_buf* db);
static const as_info_cmd* find_cmd(const char* name, size_t name_len);
static void handle_cmd(const as_info_cmd* cmd, const char* params, const as_file_handle* fd_h, cf_dyn_buf* db);

// Info commands.
static void cmd_best_practices(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_build(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_build_arch(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_build_ee_sha(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_build_os(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_build_sha(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_build_time(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_cluster_name(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_cluster_stable(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_compatibility_id(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_debug_record(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_debug_record_meta(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_digests(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_dump_cluster(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_dump_fabric(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_dump_hb(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_dump_hlc(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_dump_migrates(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_dump_rw_request_hash(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_dump_skew(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_dump_wb_summary(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_edition(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_endpoints(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_eviction_reset(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_features(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_features_key(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_get_sl(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_get_stats(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_hb_addr(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_health_outliers(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_health_stats(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_help(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_histogram(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_index_pressure(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_jem_stats(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_latencies(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_log(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_log_message(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_log_set(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_logs(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_namespace(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_namespaces(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_node(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_objects(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_partition_generation(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_partition_info(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_partitions(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_physical_devices(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_query_abort(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_query_abort_all(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_query_show(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_quiesce(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_quiesce_undo(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_rack_ids(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_racks(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_rebalance_generation(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_recluster(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_replicas(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_replicas_all(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_replicas_master(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_revive(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_roster(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_roster_set(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_sets(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_sindex(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_sindex_create(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_sindex_delete(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_sindex_exists(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_sindex_list(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_sindex_stat(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_smd_info(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_smd_show(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_statistics(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_status(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_thread_traces(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_tip(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_tip_clear(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_truncate(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_truncate_namespace(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_truncate_namespace_undo(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_truncate_undo(const char* name, const char* params, cf_dyn_buf* db);
static void cmd_version(const char* name, const char* params, cf_dyn_buf* db);

// Info command helpers.
static const char* perm_to_string(as_sec_perm perm);
static cf_ip_port bind_to_port(cf_serv_cfg* cfg, cf_sock_owner owner);
static char* access_to_string(cf_addr_list* addrs);
static void info_get_aggregated_namespace_stats(cf_dyn_buf* db);
static void info_get_namespace_info(as_namespace* ns, cf_dyn_buf* db);
static void namespace_rack_info(as_namespace* ns, cf_dyn_buf* db, uint32_t* rack_ids, uint32_t n_nodes, cf_node node_seq[], const char* tag);
static void namespace_roster_info(as_namespace* ns, cf_dyn_buf* db);
static void add_index_device_stats(as_namespace* ns, cf_dyn_buf* db);
static void add_sindex_device_stats(as_namespace* ns, cf_dyn_buf* db);
static int32_t oldest_nvme_age(const char* path);
static void add_data_stripe_stats(as_namespace* ns, cf_dyn_buf* db);
static void add_data_device_stats(as_namespace* ns, cf_dyn_buf* db);
static void find_sindex_key(const cf_vector* items, void* udata);
static void smd_show_cb(const cf_vector* items, void* udata);
static void debug_record(const char* params, cf_dyn_buf* db, bool all_data);
static void debug_dump_index(cf_dyn_buf* db, as_namespace* ns, as_record* r, bool is_orig);
static void debug_dump_pickle(cf_dyn_buf* db, as_storage_rd* rd, bool is_orig);
static void debug_dump_raw(cf_dyn_buf* db, as_storage_rd* rd, bool leave_encrypted, bool is_orig);
static void debug_dump_parsed(cf_dyn_buf* db, as_storage_rd* rd, bool all_data, bool is_orig);


//==========================================================
// Inlines & macros.
//

// TODO: This function should move elsewhere.
static inline uint64_t
get_cpu_ns(void)
{
	struct timespec ts;
	clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &ts);
	return (ts.tv_sec * 1000 * 1000 * 1000) + ts.tv_nsec;
}

// A comparison_fn_t used with qsort() - yields ascending rack-id order.
static inline int
compare_rack_nodes(const void* pa, const void* pb)
{
	uint32_t a = ((const rack_node*)pa)->rack_id;
	uint32_t b = ((const rack_node*)pb)->rack_id;

	return a > b ? 1 : (a == b ? 0 : -1);
}

static inline info_param_result
as_info_param_get_aliases(const char* params, const char* aliases[],
		char* value, int* value_len) {
	info_param_result rv;

	for (const char** select = aliases; *select != NULL; select++) {
		if ((rv = as_info_parameter_get(params, *select, value, value_len)) !=
				INFO_PARAM_FAIL_NOT_FOUND) {
			//	if (select != aliases) {
			//		cf_ticker_warning(AS_INFO, "'%s' has been deprecated - please use '%s' instead",
			//				*select, *aliases);
			//	}
			return rv;
		}
	}

	return rv;
}

static inline void
info_respond_error_argp(cf_dyn_buf* db, int num, const char* message,
		va_list argp)
{
	cf_dyn_buf_append_string(db, "ERROR:");

	if (num != AS_ERR_UNKNOWN) {
		cf_dyn_buf_append_int(db, num);
	}

	cf_dyn_buf_append_char(db, ':');
	cf_dyn_buf_append_format_va(db, message, argp);
}


//==========================================================
// Command function table.
//

static const as_info_cmd SPECS[] = {
	// Set up commands in summary (order is important for summary commands).
	{ .name="features",                .fn=cmd_features,                .in_summary=true },
	{ .name="node",                    .fn=cmd_node,                    .in_summary=true },
	{ .name="compatibility-id",        .fn=cmd_compatibility_id,        .in_summary=true },
	{ .name="edition",                 .fn=cmd_edition,                 .in_summary=true },
	{ .name="build_ee_sha",            .fn=cmd_build_ee_sha,            .in_summary=true },
	{ .name="build_sha",               .fn=cmd_build_sha,               .in_summary=true },
	{ .name="build_arch",              .fn=cmd_build_arch,              .in_summary=true },
	{ .name="build_time",              .fn=cmd_build_time,              .in_summary=true },
	{ .name="build_os",                .fn=cmd_build_os,                .in_summary=true },
	{ .name="build",                   .fn=cmd_build,                   .in_summary=true },
	{ .name="version",                 .fn=cmd_version,                 .in_summary=true },
	{ .name="statistics",              .fn=cmd_statistics,              .in_summary=true },
	{ .name="services-alumni",         .fn=as_service_list_dynamic,     .in_summary=true },
	{ .name="services",                .fn=as_service_list_dynamic,     .in_summary=true },
	{ .name="partition-generation",    .fn=cmd_partition_generation,    .in_summary=true },

	{ .name="alumni-clear-std",        .fn=as_service_list_dynamic,     .client_only=true,  .ee_only=false, .perm=PERM_NONE           },
	{ .name="alumni-tls-std",          .fn=as_service_list_dynamic,     .client_only=true,  .ee_only=true,  .perm=PERM_NONE           },
	{ .name="best-practices",          .fn=cmd_best_practices,          .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="cluster-name",            .fn=cmd_cluster_name,            .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="cluster-stable",          .fn=cmd_cluster_stable,          .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="config-get",              .fn=as_cfg_info_cmd_config_get,  .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="config-set",              .fn=as_cfg_info_cmd_config_set,  .client_only=false, .ee_only=false, .perm=PERM_SET_CONFIG     },
	{ .name="debug-record",            .fn=cmd_debug_record,            .client_only=false, .ee_only=false, .perm=PERM_RECORD_INFO    },
	{ .name="debug-record-meta",       .fn=cmd_debug_record_meta,       .client_only=false, .ee_only=false, .perm=PERM_RECORD_INFO    },
	{ .name="digests",                 .fn=cmd_digests,                 .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="dump-cluster",            .fn=cmd_dump_cluster,            .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="dump-fabric",             .fn=cmd_dump_fabric,             .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="dump-hb",                 .fn=cmd_dump_hb,                 .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="dump-hlc",                .fn=cmd_dump_hlc,                .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="dump-migrates",           .fn=cmd_dump_migrates,           .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="dump-rw",                 .fn=cmd_dump_rw_request_hash,    .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="dump-skew",               .fn=cmd_dump_skew,               .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="dump-wb-summary",         .fn=cmd_dump_wb_summary,         .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="endpoints",               .fn=cmd_endpoints,               .client_only=true,  .ee_only=false, .perm=PERM_NONE           },
	{ .name="eviction-reset",          .fn=cmd_eviction_reset,          .client_only=false, .ee_only=false, .perm=PERM_EVICT_ADMIN    },
	{ .name="feature-key",             .fn=cmd_features_key,            .client_only=false, .ee_only=true,  .perm=PERM_NONE           },
	{ .name="get-config",              .fn=as_cfg_info_cmd_config_get,  .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="get-sl",                  .fn=cmd_get_sl,                  .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="get-stats",               .fn=cmd_get_stats,               .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="health-outliers",         .fn=cmd_health_outliers,         .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="health-stats",            .fn=cmd_health_stats,            .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="help",                    .fn=cmd_help,                    .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="histogram",               .fn=cmd_histogram,               .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="index-pressure",          .fn=cmd_index_pressure,          .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="jem-stats",               .fn=cmd_jem_stats,               .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="latencies",               .fn=cmd_latencies,               .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="log",                     .fn=cmd_log,                     .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="log-message",             .fn=cmd_log_message,             .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="log-set",                 .fn=cmd_log_set,                 .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="logs",                    .fn=cmd_logs,                    .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="mcast",                   .fn=cmd_hb_addr,                 .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="mesh",                    .fn=cmd_hb_addr,                 .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="name",                    .fn=cmd_node,                    .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="namespace",               .fn=cmd_namespace,               .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="namespaces",              .fn=cmd_namespaces,              .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="objects",                 .fn=cmd_objects,                 .client_only=true,  .ee_only=false, .perm=PERM_NONE           },
	{ .name="partition-info",          .fn=cmd_partition_info,          .client_only=true,  .ee_only=false, .perm=PERM_NONE           },
	{ .name="partitions",              .fn=cmd_partitions,              .client_only=true,  .ee_only=false, .perm=PERM_NONE           },
	{ .name="peers-clear-alt",         .fn=as_service_list_command,     .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="peers-clear-std",         .fn=as_service_list_command,     .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="peers-generation",        .fn=as_service_list_dynamic,     .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="peers-tls-alt",           .fn=as_service_list_command,     .client_only=false, .ee_only=true,  .perm=PERM_NONE           },
	{ .name="peers-tls-std",           .fn=as_service_list_command,     .client_only=false, .ee_only=true,  .perm=PERM_NONE           },
	{ .name="physical-devices",        .fn=cmd_physical_devices,        .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="query-abort",             .fn=cmd_query_abort,             .client_only=false, .ee_only=false, .perm=PERM_QUERY_ADMIN    },
	{ .name="query-abort-all",         .fn=cmd_query_abort_all,         .client_only=false, .ee_only=false, .perm=PERM_QUERY_ADMIN    },
	{ .name="query-show",              .fn=cmd_query_show,              .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="quiesce",                 .fn=cmd_quiesce,                 .client_only=false, .ee_only=true,  .perm=PERM_SERVICE_CTRL   },
	{ .name="quiesce-undo",            .fn=cmd_quiesce_undo,            .client_only=false, .ee_only=true,  .perm=PERM_SERVICE_CTRL   },
	{ .name="rack-ids",                .fn=cmd_rack_ids,                .client_only=false, .ee_only=true,  .perm=PERM_NONE           },
	{ .name="racks",                   .fn=cmd_racks,                   .client_only=false, .ee_only=true,  .perm=PERM_NONE           },
	{ .name="rebalance-generation",    .fn=cmd_rebalance_generation,    .client_only=true,  .ee_only=false, .perm=PERM_NONE           },
	{ .name="recluster",               .fn=cmd_recluster,               .client_only=false, .ee_only=false, .perm=PERM_SERVICE_CTRL   },
	{ .name="replicas",                .fn=cmd_replicas,                .client_only=true,  .ee_only=false, .perm=PERM_NONE           },
	{ .name="replicas-all",            .fn=cmd_replicas_all,            .client_only=true,  .ee_only=false, .perm=PERM_NONE           },
	{ .name="replicas-master",         .fn=cmd_replicas_master,         .client_only=true,  .ee_only=false, .perm=PERM_NONE           },
	{ .name="revive",                  .fn=cmd_revive,                  .client_only=false, .ee_only=true,  .perm=PERM_SERVICE_CTRL   },
	{ .name="roster",                  .fn=cmd_roster,                  .client_only=false, .ee_only=true,  .perm=PERM_NONE           },
	{ .name="roster-set",              .fn=cmd_roster_set,              .client_only=false, .ee_only=true,  .perm=PERM_SERVICE_CTRL   },
	{ .name="service",                 .fn=as_service_list_dynamic,     .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="service-clear-alt",       .fn=as_service_list_dynamic,     .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="service-clear-std",       .fn=as_service_list_dynamic,     .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="service-tls-alt",         .fn=as_service_list_dynamic,     .client_only=false, .ee_only=true,  .perm=PERM_NONE           },
	{ .name="service-tls-std",         .fn=as_service_list_dynamic,     .client_only=false, .ee_only=true,  .perm=PERM_NONE           },
	{ .name="services-alternate",      .fn=as_service_list_dynamic,     .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="services-alumni-reset",   .fn=as_service_list_dynamic,     .client_only=false, .ee_only=false, .perm=PERM_SERVICE_CTRL   },
	{ .name="set-config",              .fn=as_cfg_info_cmd_config_set,  .client_only=false, .ee_only=false, .perm=PERM_SET_CONFIG     },
	{ .name="set-log",                 .fn=cmd_log_set,                 .client_only=false, .ee_only=false, .perm=PERM_LOGGING_CTRL   },
	{ .name="sets",                    .fn=cmd_sets,                    .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="sindex",                  .fn=cmd_sindex,                  .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="sindex-create",           .fn=cmd_sindex_create,           .client_only=false, .ee_only=false, .perm=PERM_SINDEX_ADMIN   },
	{ .name="sindex-delete",           .fn=cmd_sindex_delete,           .client_only=false, .ee_only=false, .perm=PERM_SINDEX_ADMIN   },
	{ .name="sindex-exists",           .fn=cmd_sindex_exists,           .client_only=false, .ee_only=false, .perm=PERM_SINDEX_ADMIN   },
	{ .name="sindex-list",             .fn=cmd_sindex_list,             .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="sindex-stat",             .fn=cmd_sindex_stat,             .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="smd-info",                .fn=cmd_smd_info,                .client_only=true,  .ee_only=false, .perm=PERM_NONE           },
	{ .name="smd-show",                .fn=cmd_smd_show,                .client_only=true,  .ee_only=false, .perm=PERM_NONE           },
	{ .name="status",                  .fn=cmd_status,                  .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="thread-traces",           .fn=cmd_thread_traces,           .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="tip",                     .fn=cmd_tip,                     .client_only=false, .ee_only=false, .perm=PERM_SERVICE_CTRL   },
	{ .name="tip-clear",               .fn=cmd_tip_clear,               .client_only=false, .ee_only=false, .perm=PERM_SERVICE_CTRL   },
	{ .name="truncate",                .fn=cmd_truncate,                .client_only=false, .ee_only=false, .perm=PERM_TRUNCATE       },
	{ .name="truncate-namespace",      .fn=cmd_truncate_namespace,      .client_only=false, .ee_only=false, .perm=PERM_TRUNCATE       },
	{ .name="truncate-namespace-undo", .fn=cmd_truncate_namespace_undo, .client_only=false, .ee_only=false, .perm=PERM_TRUNCATE       },
	{ .name="truncate-undo",           .fn=cmd_truncate_undo,           .client_only=false, .ee_only=false, .perm=PERM_TRUNCATE       },
	{ .name="udf-clear-cache",         .fn=udf_cask_info_clear_cache,   .client_only=true , .ee_only=false, .perm=PERM_UDF_ADMIN      },
	{ .name="udf-get",                 .fn=udf_cask_info_get,           .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="udf-list",                .fn=udf_cask_info_list,          .client_only=false, .ee_only=false, .perm=PERM_NONE           },
	{ .name="udf-put",                 .fn=udf_cask_info_put,           .client_only=false, .ee_only=false, .perm=PERM_UDF_ADMIN      },
	{ .name="udf-remove",              .fn=udf_cask_info_remove,        .client_only=false, .ee_only=false, .perm=PERM_UDF_ADMIN      },
	{ .name="xdr-dc-state",            .fn=as_xdr_dc_state,             .client_only=false, .ee_only=true,  .perm=PERM_NONE           },
	{ .name="xdr-get-filter",          .fn=as_xdr_get_filter,           .client_only=false, .ee_only=true,  .perm=PERM_NONE           },
	{ .name="xdr-set-filter",          .fn=as_xdr_set_filter,           .client_only=false, .ee_only=true,  .perm=PERM_XDR_SET_FILTER }
};

static const uint32_t N_SPECS = sizeof(SPECS) / sizeof(SPECS[0]);


//==========================================================
// Public API.
//

void
as_info_init()
{
	cf_queue_init(&g_info_work_q, sizeof(as_info_transaction), 64, true);
	g_specs = cf_shash_create(cf_shash_fn_zstr, CMD_NAME_MAX_LEN,
			sizeof(as_info_cmd*), N_SPECS * 2, false);

	const as_info_cmd* prior = NULL;

	for (uint32_t i = 0; i < N_SPECS; i++) {
		const as_info_cmd* cur = &SPECS[i];
		char key[CMD_NAME_MAX_LEN] = { 0 };
		uint32_t name_len = strlen(cur->name);

		cf_assert(name_len < CMD_NAME_MAX_LEN, AS_INFO, "cmd name '%s' len exceeds %u",
				cur->name, CMD_NAME_MAX_LEN);

		cf_assert(cur->perm == PERM_NONE || ! cur->in_summary || ! cur->client_only,
				AS_INFO, "cmd name '%s' in summary with illegal permission",
				cur->name);

		memcpy(key, cur->name, name_len);

		if (cf_shash_put_unique(g_specs, key, &cur) != CF_SHASH_OK) {
			cf_crash(AS_INFO, "duplicate command spec found - '%s'", cur->name);
		}

		if (prior != NULL) {
			if (! prior->in_summary) {
				cf_assert(strcmp(prior->name, cur->name) < 0, AS_INFO, "specs not sorted or name not unique - '%s' >= '%s'",
						prior->name, cur->name);
			}
		}

		prior = cur;
	}

	for (uint32_t i = 0; i < g_config.n_info_threads; i++) {
		cf_thread_create_transient(run_info, NULL);
	}
}

void
as_info(as_info_transaction* it)
{
	cf_queue_push(&g_info_work_q, it);
}

// Commands have the form:
// cmd:param=value;param=value
//
// @return  0 : success
//         -1 : parameter not found
//         -2 : parameter found but value is too long
info_param_result
as_info_parameter_get(const char* param_str, const char* param, char* value,
		int* value_len)
{
	cf_detail(AS_INFO, "parameter get: paramstr '%s' seeking param '%s'",
			param_str, param);

	const char* c = param_str;
	const char* tok = param_str;
	int param_len = strlen(param);

	while (*c != '\0') {
		if (*c == '=') {
			if (param_len == (c - tok) && memcmp(tok, param, param_len) == 0) {
				c++;
				tok = c;

				while (*c != '\0' && *c != ';') {
					c++;
				}

				if (*value_len <= c - tok) {
					// The found value is too long.
					return INFO_PARAM_FAIL_TOO_LONG;
				}

				*value_len = c - tok;
				memcpy(value, tok, *value_len);
				value[*value_len] = 0;

				return INFO_PARAM_OK;
			}
			c++;
		}
		else if (*c == ';') {
			c++;
			tok = c;
		}
		else {
			c++;
		}
	}

	return INFO_PARAM_FAIL_NOT_FOUND;
}

info_param_result
as_info_param_get_namespace_id(const char* params, char* value, int* value_len)
{
	const char* aliases[] = { "namespace", "id", NULL };

	return as_info_param_get_aliases(params, aliases, value, value_len);
}

info_param_result
as_info_param_get_namespace_ns(const char* params, char* value, int* value_len)
{
	const char* aliases[] = { "namespace", "ns", NULL };

	return as_info_param_get_aliases(params, aliases, value, value_len);
}

info_param_result
as_info_param_get_namespace(const char* params, char* value, int* value_len)
{
	return as_info_parameter_get(params, "namespace", value, value_len);
}

bool
as_info_required_param_is_ok(cf_dyn_buf* db, const char* param,
		const char* value, info_param_result result)
{
	result = as_info_optional_param_is_ok(db, param, value, result);

	if (result != INFO_PARAM_OK_NOT_FOUND) {
		return result == INFO_PARAM_OK;
	}

	int err = strcmp("namespace", param) == 0 ?
			AS_ERR_NAMESPACE : AS_ERR_PARAMETER;

	cf_warning(AS_INFO, "missing '%s'", param);
	as_info_respond_error(db, err, "missing '%s'", param);

	return false;
}

info_param_result
as_info_optional_param_is_ok(cf_dyn_buf* db, const char* param,
		const char* value, info_param_result result)
{
	if (result == INFO_PARAM_FAIL_NOT_FOUND) {
		return INFO_PARAM_OK_NOT_FOUND;
	}

	if (result == INFO_PARAM_OK && value[0] != '\0') {
		return INFO_PARAM_OK;
	}

	int err = strcmp("namespace", param) == 0 ?
			AS_ERR_NAMESPACE : AS_ERR_PARAMETER;

	switch (result) {
	case INFO_PARAM_OK:
		cf_warning(AS_INFO, "missing '%s' value", param);
		as_info_respond_error(db, err, "missing '%s' value", param);
		return INFO_PARAM_FAIL_REPLIED;
	case INFO_PARAM_FAIL_TOO_LONG:
		cf_warning(AS_INFO, "'%s' value too long", param);
		as_info_respond_error(db, err, "'%s' value too long", param);
		return INFO_PARAM_FAIL_REPLIED;
	default:
		cf_crash(AS_INFO, "unexpected result %d", result);
	}

	return INFO_PARAM_FAIL_REPLIED; // unreachable
}

bool
info_param_required_local_namespace_is_ok(cf_dyn_buf* db, char* value,
		as_namespace** ns, info_param_result result)
{
	if (! as_info_required_param_is_ok(db, "namespace", value, result)) {
		return false;
	}

	if (result != INFO_PARAM_OK) {
		return false;
	}

	if ((*ns = as_namespace_get_byname(value)) == NULL) {
		cf_warning(AS_INFO, "namespace '%s' not found on node", value);
		as_info_respond_error(db, AS_ERR_NAMESPACE,
				"namespace not found on node");
		return false;
	}

	return true;
}

info_param_result
info_param_optional_local_namespace_is_ok(cf_dyn_buf* db, char* value,
		as_namespace** ns, info_param_result result)
{
	result = as_info_optional_param_is_ok(db, "namespace", value, result);

	if (result != INFO_PARAM_OK) {
		return result;
	}

	if ((*ns = as_namespace_get_byname(value)) == NULL) {
		cf_warning(AS_INFO, "namespace '%s' not found on node", value);
		as_info_respond_error(db, AS_ERR_NAMESPACE,
				"namespace not found on node");
		return INFO_PARAM_FAIL_REPLIED;
	}

	return INFO_PARAM_OK;
}

void
as_info_buffer(uint8_t* req_buf, size_t req_buf_len, cf_dyn_buf* rsp)
{
	if (req_buf_len == 0) {
		info_summary(rsp);
	}
	else {
		handle_cmds((char*)req_buf, req_buf_len, NULL, rsp);
	}
}

// Called via info command. Caller has sanity-checked n_threads.
void
as_info_set_num_info_threads(uint32_t n_threads)
{
	if (g_config.n_info_threads > n_threads) {
		// Decrease the number of info threads to n_threads.
		while (g_config.n_info_threads > n_threads) {
			as_info_transaction death_msg = { 0 };

			// Send terminator (NULL message).
			as_info(&death_msg);
			g_config.n_info_threads--;
		}
	}
	else {
		// Increase the number of info threads to n_threads.
		while (g_config.n_info_threads < n_threads) {
			cf_thread_create_transient(run_info, NULL);
			g_config.n_info_threads++;
		}
	}
}

void
as_info_respond_error(cf_dyn_buf* db, int num, const char* message, ...)
{
	va_list argp;

	va_start(argp, message);
	info_respond_error_argp(db, num, message, argp);
	va_end(argp);
}

bool
as_info_respond_enterprise_only(cf_dyn_buf* db)
{
	if (as_error_enterprise_only()) {
		as_info_respond_error(db, AS_ERR_ENTERPRISE_ONLY, "enterprise only");
		return true;
	}

	return false;
}

void
as_info_respond_ok(cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db, "ok");
}

char*
as_info_bind_to_string(const cf_serv_cfg* cfg, cf_sock_owner owner)
{
	cf_dyn_buf_define_size(db, 2500);
	uint32_t count = 0;

	for (uint32_t i = 0; i < cfg->n_cfgs; ++i) {
		if (cfg->cfgs[i].owner != owner) {
			continue;
		}

		if (count > 0) {
			cf_dyn_buf_append_char(&db, ',');
		}

		cf_dyn_buf_append_string(&db, cf_ip_addr_print(&cfg->cfgs[i].addr));
		++count;
	}

	char* string = cf_dyn_buf_strdup(&db);

	cf_dyn_buf_free(&db);

	return string != NULL ? string : cf_strdup("null");
}

// Return the number of pending Info requests in the queue.
uint32_t
as_info_queue_get_size()
{
	return cf_queue_sz(&g_info_work_q);
}

// TODO: This function should move elsewhere.
// Called only from the ticker thread.
uint32_t
process_cpu(void)
{
	static uint64_t prev = 0;
	static uint64_t prev_cpu = 0;

	uint64_t now = cf_getns();
	uint64_t now_cpu = get_cpu_ns();

	if (prev != 0) {
		uint64_t delta = now - prev;
		uint64_t delta_cpu = now_cpu - prev_cpu;

		g_process_cpu_pct = (uint32_t)(delta_cpu * 100 / delta);
	}

	prev = now;
	prev_cpu = now_cpu;

	return g_process_cpu_pct;
}

// TODO: This function should move elsewhere.
// Called only from the ticker thread.
void
sys_cpu_info(uint32_t* user_pct, uint32_t* kernel_pct)
{
	if (user_pct != NULL) {
		*user_pct = g_user_cpu_pct;
	}

	if (kernel_pct != NULL) {
		*kernel_pct = g_kernel_cpu_pct;
	}

	FILE* fh = fopen("/proc/stat", "r");

	if (fh == NULL) {
		cf_warning(AS_INFO, "failed to open /proc/stat: %d", errno);
		return;
	}

	uint64_t user;
	uint64_t nice;
	uint64_t kernel;
	uint64_t idle;

	if (fscanf(fh, "cpu %lu %lu %lu %lu", &user, &nice, &kernel, &idle) != 4) {
		cf_warning(AS_INFO, "can't parse /proc/stat");
		fclose(fh);
		return;
	}

	fclose(fh);

	static bool running = false;
	static uint64_t prev_user = 0;
	static uint64_t prev_nice = 0;
	static uint64_t prev_kernel = 0;
	static uint64_t prev_idle = 0;

	if (running) {
		uint32_t delta_user = (uint32_t)(user - prev_user);
		uint32_t delta_nice = (uint32_t)(nice - prev_nice);
		uint32_t delta_kernel = (uint32_t)(kernel - prev_kernel);
		uint32_t delta_idle = (uint32_t)(idle - prev_idle);

		uint32_t total = delta_user + delta_nice + delta_kernel + delta_idle;
		uint32_t n_cpus = cf_topo_count_cpus();

		g_user_cpu_pct = total == 0 ?
				0 : (delta_user + delta_nice) * 100 * n_cpus / total;
		g_kernel_cpu_pct = total == 0 ?
				0 : delta_kernel * 100 * n_cpus / total;
	}

	running = true;
	prev_user = user;
	prev_nice = nice;
	prev_kernel = kernel;
	prev_idle = idle;

	if (user_pct != NULL) {
		*user_pct = g_user_cpu_pct;
	}

	if (kernel_pct != NULL) {
		*kernel_pct = g_kernel_cpu_pct;
	}
}

// TODO: This function should move elsewhere.
void
sys_mem_info(uint64_t* free_mem_kbytes, uint32_t* free_mem_pct,
		uint64_t* thp_mem_kbytes)
{
	*free_mem_kbytes = 0;
	*free_mem_pct = 0;
	*thp_mem_kbytes = 0;

	int32_t fd = open("/proc/meminfo", O_RDONLY);

	if (fd < 0) {
		cf_warning(AS_INFO, "failed to open /proc/meminfo: %d", errno);
		return;
	}

	char buf[4096] = { 0 };
	size_t limit = sizeof(buf);
	size_t total = 0;

	while (total < limit) {
		ssize_t len = read(fd, buf + total, limit - total);

		if (len < 0) {
			cf_warning(AS_INFO, "couldn't read /proc/meminfo: %d", errno);
			close(fd);
			return;
		}

		if (len == 0) {
			break; // EOF
		}

		total += (size_t)len;
	}

	close(fd);

	if (total == limit) {
		cf_warning(AS_INFO, "/proc/meminfo exceeds %zu bytes", limit);
		return;
	}

	uint64_t mem_total = 0;
	uint64_t mem_available = 0;
	uint64_t anon_huge_pages = 0;

	char* cur = buf;
	char* save_ptr = NULL;

	// We split each line into two fields separated by ':'. strtoul() will
	// safely ignore the spaces and 'kB' (if present).
	while (true) {
		char* name_tok = strtok_r(cur, ":", &save_ptr);

		if (name_tok == NULL) {
			break; // no more lines
		}

		cur = NULL; // all except first name_tok use NULL

		char* value_tok = strtok_r(NULL, "\r\n", &save_ptr);

		if (value_tok == NULL) {
			cf_warning(AS_INFO, "/proc/meminfo line missing value token");
			return;
		}

		if (strcmp(name_tok, "MemTotal") == 0) {
			mem_total = strtoul(value_tok, NULL, 0);
		}
		else if (strcmp(name_tok, "MemAvailable") == 0) {
			mem_available = strtoul(value_tok, NULL, 0);
		}
		else if (strcmp(name_tok, "AnonHugePages") == 0) {
			anon_huge_pages = strtoul(value_tok, NULL, 0);
		}
	}

	*free_mem_kbytes = mem_available;
	*free_mem_pct = mem_total == 0 ? 0 : (mem_available * 100) / mem_total;
	*thp_mem_kbytes = anon_huge_pages;
}


//==========================================================
// Local helpers - info execution.
//

static void*
run_info(void* arg)
{
	(void)arg;
	as_info_transaction it;

	while (cf_queue_pop(&g_info_work_q, &it, CF_QUEUE_FOREVER) == CF_QUEUE_OK) {
		if (it.fd_h == NULL) {
			break; // termination signal
		}

		bool timeout = cf_getns() > it.start_time + g_config.info_max_ns;

		as_file_handle* fd_h = it.fd_h;
		as_proto* proto = it.proto;

		cf_detail(AS_INFO_COMMAND, "client %s command%s '%.*s'", fd_h->client,
				timeout ? " (timed out)" : "", (int)proto->sz, proto->body);

		// Allocate an output stack buffer large enough to avoid resizing.
		cf_dyn_buf_define_size(db, 128 * 1024);

		// Leave space for the response header.
		cf_dyn_buf_reserve(&db, sizeof(as_proto), NULL);

		if (timeout) {
			as_info_respond_error(&db, AS_ERR_TIMEOUT, "timeout");
			cf_dyn_buf_append_char(&db, EOL);
		}
		else if (authenticate(fd_h, &db)) {
			if (proto->sz == 0) {
				info_summary(&db); // no commands specified
			}
			else {
				handle_cmds((char*)proto->body, proto->sz, fd_h, &db);
			}
		}

		cf_free(proto);

		// Write response header into reserved space.
		as_proto* header = (as_proto*)db.buf;

		*header = (as_proto){
				.version = PROTO_VERSION,
				.type = PROTO_TYPE_INFO,
				.sz = db.used_sz - sizeof(as_proto)
		};

		as_proto_swap(header);

		if (cf_socket_send_all(&fd_h->sock, db.buf, db.used_sz,
				MSG_NOSIGNAL, CF_SOCKET_TIMEOUT) == 0) {
			as_end_of_transaction_ok(fd_h);
		}
		else {
			cf_info(AS_INFO, "error sending to %s - fd %d sz %zu %s",
					fd_h->client, CSFD(&fd_h->sock), db.used_sz,
					cf_strerror(errno));
			as_end_of_transaction_force_close(fd_h);
		}

		cf_dyn_buf_free(&db);

		if (timeout) {
			as_incr_uint64(&g_stats.info_timeout);
		}
		else {
			G_HIST_INSERT_DATA_POINT(info_hist, it.start_time);
			as_incr_uint64(&g_stats.info_complete);
		}
	}

	return NULL;
}

static bool
authenticate(const as_file_handle* fd_h, cf_dyn_buf* db)
{
	uint8_t auth_result = as_security_check_auth(fd_h);

	if (auth_result == AS_OK) {
		return true;
	}

	as_security_log(fd_h, auth_result, PERM_NONE, "info request", NULL);

	append_security_error(db, auth_result, PERM_NONE);
	cf_dyn_buf_append_char(db, EOL);

	return false;
}

static void
info_summary(cf_dyn_buf* db)
{
	for (uint32_t i = 0; i < N_SPECS; i++) {
		const as_info_cmd* cmd = &SPECS[i];

		if (cmd->in_summary == true) {
			cf_dyn_buf_append_string(db, cmd->name);
			cf_dyn_buf_append_char(db, SEP);
			cmd->fn(cmd->name, "", db);
			cf_dyn_buf_append_char(db, EOL);
		}
	}
}

static void
handle_cmds(char* buf, size_t buf_sz, const as_file_handle* fd_h,
		cf_dyn_buf* db)
{
	char* end = buf + buf_sz;
	char* at = buf;

	for (char* c = buf; c < end; c++) {
		if (*c != EOL) {
			continue;
		}

		*c = '\0'; // null terminate command

		char* cmd_str = at;

		// Return exact command string received from client.
		cf_dyn_buf_append_string(db, cmd_str);
		cf_dyn_buf_append_char(db, SEP);

		const as_info_cmd* cmd = parse_cmd(&at, db);

		if (cmd != NULL) {
			handle_cmd(cmd, at, fd_h, db);
		}
		else {
			cf_warning(AS_INFO, "unrecognized command - '%s'", cmd_str);
			as_info_respond_error(db, AS_ERR_PARAMETER, "unrecognized command");
		}

		cf_dyn_buf_append_char(db, EOL);
		at = c + 1;
	}
}

static const as_info_cmd*
parse_cmd(char** cmd_str_p, cf_dyn_buf* db)
{
	// Command may have the following syntax:
	//   cmd-name/arg0/arg1/...:key0=val0;key1=val1/...;

	char* cmd_str = *cmd_str_p;
	char* at = strchr(cmd_str, ARG_SEP);
	char* args_kwargs_at = strchr(cmd_str, ARG_KWARG_SEP);
	char* kwargs_at = strchr(cmd_str, KWARG_SEP);

	if (at == NULL || (args_kwargs_at != NULL && args_kwargs_at < at)) {
		// Ignore args separator after args/kwargs separator.
		at = args_kwargs_at;
	}

	if (at == NULL || (kwargs_at != NULL && kwargs_at < at)) {
		// Allow for a command with a trailing ';'.
		at = kwargs_at;
	}

	size_t name_len;

	if (at != NULL) {
		name_len = at - cmd_str;
		*cmd_str_p = at + 1;
	}
	else {
		name_len = strlen(*cmd_str_p);
		*cmd_str_p = *cmd_str_p + name_len;
	}

	return find_cmd(cmd_str, name_len);
}

static const as_info_cmd*
find_cmd(const char* name, size_t name_len)
{
	if (name_len >= CMD_NAME_MAX_LEN) {
		return NULL;
	}

	char key[CMD_NAME_MAX_LEN] = { 0 };
	as_info_cmd* res = NULL;

	memcpy(key, name, name_len);
	cf_shash_get(g_specs, key, &res);

	return res;
}

static void
handle_cmd(const as_info_cmd* cmd, const char* params,
		const as_file_handle* fd_h, cf_dyn_buf* db)
{
	if (cmd->ee_only && as_info_respond_enterprise_only(db)) {
		return;
	}

	as_sec_perm perm = cmd->perm;

	// Already authenticated so ignore PERM_NONE.
	if (perm != PERM_NONE) {
		uint8_t result = as_security_check_info_cmd(fd_h, cmd->name, params,
				perm);

		as_security_log(fd_h, result, perm, cmd->name, params);

		if (result != AS_OK) {
			append_security_error(db, result, perm);
			return;
		}
	}

	cmd->fn(cmd->name, params, db);
}

static void
append_security_error(cf_dyn_buf* db, uint32_t result, as_sec_perm perm)
{
	switch (result) {
	case AS_SEC_ERR_NOT_AUTHENTICATED:
		as_info_respond_error(db, result, "not authenticated");
		return;
	case AS_SEC_ERR_ROLE_VIOLATION:
		as_info_respond_error(db, result,
				"role violation - requires permission '%s'",
				perm_to_string(perm));
		return;
	default:
		break;
	}

	as_info_respond_error(db, result, "unexpected security result");
}


//==========================================================
// Local helpers - info commands.
//

static void
cmd_best_practices(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db, "failed_best_practices=");

	if (g_bad_practices.used_sz == 0) {
		cf_dyn_buf_append_string(db, "none");
	}
	else {
		cf_dyn_buf_append_buf(db, g_bad_practices.buf, g_bad_practices.used_sz);
	}
}

static void
cmd_build(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db, aerospike_build_id);
}

static void
cmd_build_arch(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db, aerospike_build_arch);
}

static void
cmd_build_ee_sha(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db, aerospike_build_ee_sha);
}

static void
cmd_build_os(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db, aerospike_build_os);
}

static void
cmd_build_sha(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db, aerospike_build_sha);
}

static void
cmd_build_time(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db, aerospike_build_time);
}

static void
cmd_cluster_name(const char* name, const char* params, cf_dyn_buf* db)
{
	char cluster_name[AS_CLUSTER_NAME_SZ];

	as_config_cluster_name_get(cluster_name);
	cf_dyn_buf_append_string(db, cluster_name);
}

static void
cmd_cluster_stable(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// cluster-stable:[size=<target-size>];[ignore-migrations=<bool>];[namespace=<namespace-name>]

	uint64_t begin_cluster_key = as_exchange_cluster_key();

	if (! as_partition_balance_are_migrations_allowed()) {
		as_info_respond_error(db, AS_ERR_CLUSTER_KEY_MISMATCH,
				"unstable cluster");
		return;
	}

	char size_str[4]; // max cluster size is 256
	int size_str_len = (int)sizeof(size_str);
	info_param_result rv = as_info_parameter_get(params, "size", size_str,
			&size_str_len);

	rv = as_info_optional_param_is_ok(db, "size", size_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		uint32_t target_size;

		if (cf_str_atoi_u32(size_str, &target_size) != 0) {
			cf_warning(AS_INFO, "non-integer size parameter");
			as_info_respond_error(db, AS_ERR_PARAMETER, "bad size");
			return;
		}

		if (target_size != as_exchange_cluster_size()) {
			as_info_respond_error(db, AS_ERR_CLUSTER_KEY_MISMATCH,
					"cluster not specified size");
			return;
		}
	}

	bool ignore_migrations = false;

	char ignore_migrations_str[6];
	int ignore_migrations_str_len = (int)sizeof(ignore_migrations_str);

	rv = as_info_parameter_get(params, "ignore-migrations",
			ignore_migrations_str, &ignore_migrations_str_len);
	rv = as_info_optional_param_is_ok(db, "ignore-migrations",
			ignore_migrations_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		if (strcmp(ignore_migrations_str, "true") == 0) {
			ignore_migrations = true;
		}
		else if (strcmp(ignore_migrations_str, "false") == 0) {
			ignore_migrations = false;
		}
		else {
			cf_warning(AS_INFO, "ignore-migrations value invalid");
			as_info_respond_error(db, AS_ERR_PARAMETER,
					"bad ignore migrations");
			return;
		}
	}

	if (! ignore_migrations) {
		char ns_name[AS_ID_NAMESPACE_SZ] = { 0 };
		int ns_name_len = (int)sizeof(ns_name);
		as_namespace* ns = NULL;

		rv = as_info_param_get_namespace(params, ns_name, &ns_name_len);
		rv = info_param_optional_local_namespace_is_ok(db, ns_name, &ns, rv);

		if (rv == INFO_PARAM_FAIL_REPLIED) {
			return;
		}

		uint32_t n_namespaces = g_config.n_namespaces;
		as_namespace** namespaces = g_config.namespaces;

		if (rv == INFO_PARAM_OK) {
			// Ensure migrations are complete for the requested namespace only.
			n_namespaces = 1;
			namespaces = &namespaces[ns->ix];
		}

		for (uint32_t ns_ix = 0; ns_ix < n_namespaces; ns_ix++) {
			ns = namespaces[ns_ix];

			if (ns->migrate_tx_partitions_remaining +
					ns->migrate_rx_partitions_remaining +
					ns->n_unavailable_partitions +
					ns->n_dead_partitions != 0) {
				as_info_respond_error(db, AS_ERR_CLUSTER_KEY_MISMATCH,
						"unstable cluster");
				return;
			}
		}
	}

	if (begin_cluster_key != as_exchange_cluster_key()) {
		// Verify that the cluster didn't change while during the collection.
		as_info_respond_error(db, AS_ERR_CLUSTER_KEY_MISMATCH,
				"unstable cluster");
	}

	cf_dyn_buf_append_uint64_x(db, begin_cluster_key);
}

static void
cmd_compatibility_id(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_uint32(db, AS_EXCHANGE_COMPATIBILITY_ID);
}

static void
cmd_debug_record(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// debug-record:namespace=<ns-name>;keyd=<hex-digest>[;mode=<mode>]
	// where <mode> is one of: pickle or raw or raw-encrypted

	debug_record(params, db, true);
}

static void
cmd_debug_record_meta(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: debug-record-meta:namespace=<ns-name>;keyd=<hex-digest>

	debug_record(params, db, false);
}

static void
cmd_digests(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db, "RIPEMD160");
}

static void
cmd_dump_cluster(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// dump-cluster:{verbose=<opt>}
	// where the "verbose" argument is optional and <opt> is one of:
	// {"true" | "false"} and defaults to "false".

	bool verbose = false;
	char param_str[100];
	int param_str_len = sizeof(param_str);
	info_param_result rv  = as_info_parameter_get(params, "verbose", param_str,
			&param_str_len);

	rv = as_info_optional_param_is_ok(db, "verbose", param_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		if (! strcmp(param_str, "true")) {
			verbose = true;
		}
		else if (! strcmp(param_str, "false")) {
			verbose = false;
		}
		else {
			cf_warning(AS_INFO, "The '%s:' command argument 'verbose' value must be one of {'true', 'false'}, not '%s'",
					name, param_str);
			as_info_respond_error(db, AS_ERR_PARAMETER,
					"'verbose' may be either 'true' or 'false'");
			return;
		}
	}

	as_clustering_dump(verbose);
	as_exchange_dump(verbose);
	as_info_respond_ok(db);
}

static void
cmd_dump_fabric(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// dump-fabric:{verbose=<opt>}
	// where the "verbose" argument is optional] and the <opt> is one of:
	// {"true" | "false"} and defaults to "false".

	bool verbose = false;
	char param_str[100];
	int param_str_len = sizeof(param_str);
	info_param_result rv = as_info_parameter_get(params, "verbose", param_str,
			&param_str_len);

	rv = as_info_optional_param_is_ok(db, "verbose", param_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		if (! strcmp(param_str, "true")) {
			verbose = true;
		}
		else if (! strcmp(param_str, "false")) {
			verbose = false;
		}
		else {
			cf_warning(AS_INFO, "The '%s:' command argument 'verbose' value must be one of {'true', 'false'}, not '%s'",
					name, param_str);
			as_info_respond_error(db, AS_ERR_PARAMETER,
					"'verbose' may be either 'true' or 'false'");
			return;
		}
	}

	as_fabric_dump(verbose);
	as_info_respond_ok(db);
}

static void
cmd_dump_hb(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: dump-hb:{verbose=<opt>}
	// where the "verbose" argument is optional and the <opt> is one of:
	// {"true" | "false"} and defaults to "false".

	bool verbose = false;
	char param_str[100];
	int param_str_len = sizeof(param_str);
	info_param_result rv = as_info_parameter_get(params, "verbose", param_str,
			&param_str_len);

	rv = as_info_optional_param_is_ok(db, "verbose", param_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		if (! strcmp(param_str, "true")) {
			verbose = true;
		}
		else if (! strcmp(param_str, "false")) {
			verbose = false;
		}
		else {
			cf_warning(AS_INFO, "The '%s:' command argument 'verbose' value must be one of {'true', 'false'}, not '%s'",
					name, param_str);
			as_info_respond_error(db, AS_ERR_PARAMETER,
					"'verbose' may be either 'true' or 'false'");
			return;
		}
	}

	as_hb_dump(verbose);
	as_info_respond_ok(db);
}

static void
cmd_dump_hlc(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// dump-hlc:{verbose=<opt>}
	// where the "verbose" argument is optional] and the <opt> is one of:
	// {"true" | "false"} and defaults to "false".

	bool verbose = false;
	char param_str[100];
	int param_str_len = sizeof(param_str);
	info_param_result rv = as_info_parameter_get(params, "verbose", param_str,
			&param_str_len);

	rv = as_info_optional_param_is_ok(db, "verbose", param_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		if (! strcmp(param_str, "true")) {
			verbose = true;
		}
		else if (! strcmp(param_str, "false")) {
			verbose = false;
		}
		else {
			cf_warning(AS_INFO, "The '%s:' command argument 'verbose' value must be one of {'true', 'false'}, not '%s'",
					name, param_str);
			as_info_respond_error(db, AS_ERR_PARAMETER,
					"'verbose' may be either 'true' or 'false'");
			return;
		}
	}

	as_hlc_dump(verbose);
	as_info_respond_ok(db);
}

static void
cmd_dump_migrates(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// dump-migrates:{verbose=<opt>}
	// where <opt> is one of:  {"true" | "false"} and defaults to "false".

	bool verbose = false;
	char param_str[100];
	int param_str_len = sizeof(param_str);
	info_param_result rv = as_info_parameter_get(params, "verbose", param_str,
			&param_str_len);

	rv = as_info_optional_param_is_ok(db, "verbose", param_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		if (! strcmp(param_str, "true")) {
			verbose = true;
		}
		else if (! strcmp(param_str, "false")) {
			verbose = false;
		}
		else {
			cf_warning(AS_INFO, "The '%s:' command argument 'verbose' value must be one of {'true', 'false'}, not '%s'",
					name, param_str);
			as_info_respond_error(db, AS_ERR_PARAMETER,
					"'verbose' may be either 'true' or 'false'");
			return;
		}
	}

	as_migrate_dump(verbose);
	as_info_respond_ok(db);
}

static void
cmd_dump_rw_request_hash(const char* name, const char* params, cf_dyn_buf* db)
{
	rw_request_hash_dump();
	as_info_respond_ok(db);
}

static void
cmd_dump_skew(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: dump-skew:

	cf_debug(AS_INFO, "dump-skew command received: params %s", params);

	as_skew_monitor_dump();
	as_info_respond_ok(db);
}

static void
cmd_dump_wb_summary(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: dump-wb-summary:ns=<Namespace>
	// where <Namespace> is the name of an existing namespace.

	as_namespace* ns;
	char param_str[100];
	int param_str_len = sizeof(param_str);
	info_param_result rv = as_info_param_get_namespace_ns(params, param_str,
			&param_str_len);

	if (! info_param_required_local_namespace_is_ok(db, param_str, &ns, rv)) {
		return;
	}

	as_storage_dump_wb_summary(ns);
	as_info_respond_ok(db);
}

static void
cmd_edition(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db, aerospike_build_type);
}

static void
cmd_endpoints(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_ip_port port = bind_to_port(&g_service_bind, CF_SOCK_OWNER_SERVICE);
	info_append_int(db, "service.port", port);

	char* string = as_info_bind_to_string(&g_service_bind,
			CF_SOCK_OWNER_SERVICE);

	info_append_string(db, "service.addresses", string);
	cf_free(string);

	info_append_int(db, "service.access-port", g_access.service.port);

	string = access_to_string(&g_access.service.addrs);
	info_append_string(db, "service.access-addresses", string);
	cf_free(string);

	info_append_int(db, "service.alternate-access-port", g_access.alt_service.port);

	string = access_to_string(&g_access.alt_service.addrs);
	info_append_string(db, "service.alternate-access-addresses", string);
	cf_free(string);

	port = bind_to_port(&g_service_bind, CF_SOCK_OWNER_SERVICE_TLS);
	info_append_int(db, "service.tls-port", port);

	string = as_info_bind_to_string(&g_service_bind, CF_SOCK_OWNER_SERVICE_TLS);
	info_append_string(db, "service.tls-addresses", string);
	cf_free(string);

	info_append_int(db, "service.tls-access-port", g_access.tls_service.port);

	string = access_to_string(&g_access.tls_service.addrs);
	info_append_string(db, "service.tls-access-addresses", string);
	cf_free(string);

	info_append_int(db, "service.tls-alternate-access-port",
			g_access.alt_tls_service.port);

	string = access_to_string(&g_access.alt_tls_service.addrs);
	info_append_string(db, "service.tls-alternate-access-addresses", string);
	cf_free(string);

	as_hb_info_endpoints_get(db);

	port = bind_to_port(&g_fabric_bind, CF_SOCK_OWNER_FABRIC);
	info_append_int(db, "fabric.port", port);

	string = as_info_bind_to_string(&g_fabric_bind, CF_SOCK_OWNER_FABRIC);
	info_append_string(db, "fabric.addresses", string);
	cf_free(string);

	port = bind_to_port(&g_fabric_bind, CF_SOCK_OWNER_FABRIC_TLS);
	info_append_int(db, "fabric.tls-port", port);

	string = as_info_bind_to_string(&g_fabric_bind, CF_SOCK_OWNER_FABRIC_TLS);
	info_append_string(db, "fabric.tls-addresses", string);
	cf_free(string);

	as_fabric_info_peer_endpoints_get(db);

	info_append_int(db, "info.port", g_info_port);

	string = as_info_bind_to_string(&g_info_bind, CF_SOCK_OWNER_INFO);
	info_append_string(db, "info.addresses", string);
	cf_free(string);

	cf_dyn_buf_chomp(db);
}

static void
cmd_eviction_reset(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// eviction-reset:namespace=<ns-name>[;ttl=<seconds-from-now>]
	// where no ttl means delete the SMD evict-void-time.

	// Get the namespace name.

	char ns_name[AS_ID_NAMESPACE_SZ];
	int ns_name_len = (int)sizeof(ns_name);
	info_param_result ns_rv = as_info_param_get_namespace(params, ns_name,
			&ns_name_len);

	if (! as_info_required_param_is_ok(db, "namespace", ns_name, ns_rv)) {
		return;
	}

	// Get the TTL if there is one.

	char ttl_str[12]; // allow decimal, hex or octal in C constant format
	int ttl_str_len = (int)sizeof(ttl_str);
	info_param_result ttl_rv = as_info_parameter_get(params, "ttl", ttl_str,
			&ttl_str_len);

	ttl_rv = as_info_optional_param_is_ok(db, "ttl", ttl_str, ttl_rv);

	if (ttl_rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	// Issue the eviction-reset command.

	as_nsup_eviction_reset_cmd(ns_name, ttl_rv == INFO_PARAM_OK ?
			ttl_str : NULL, db);
}

static void
cmd_features(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db,
			"batch-any;batch-index;blob-bits;"
			"cdt-index;cdt-list;cdt-map;cluster-stable;"
			"float;"
			"geo;"
			"sindex-exists;"
			"peers;pipelining;pquery;pscans;"
			"query-show;"
			"relaxed-sc;replicas;replicas-all;replicas-master;replicas-max;"
			"truncate-namespace;"
			"udf"
			);
	cf_dyn_buf_append_string(db, aerospike_build_features);
}

static void
cmd_features_key(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_string(db, as_features_info());
}

static void
cmd_get_sl(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: get-sl:
	as_exchange_info_get_succession(db);
}

static void
cmd_get_stats(const char* name, const char* params, cf_dyn_buf* db)
{
	char context[1024];
	int context_len = sizeof(context);
	info_param_result rv = as_info_parameter_get(params, "context", context,
			&context_len);

	if (! as_info_required_param_is_ok(db, "context", context, rv)) {
		return;
	}

	if (strcmp(context, "xdr") == 0) {
		as_xdr_get_stats(params, db);
	}
	else {
		as_info_respond_error(db, AS_ERR_PARAMETER, "unknown context");
	}
}

static void
cmd_hb_addr(const char* name, const char* params, cf_dyn_buf* db)
{
	as_hb_mode hb_mode;
	char addr_port_str[1024];

	as_hb_info_listen_addr_get(&hb_mode, addr_port_str, sizeof(addr_port_str));

	if (hb_mode == AS_HB_MODE_MESH) {
		if (strcmp(name, "mesh") != 0) {
			as_info_respond_error(db, AS_ERR_PARAMETER,
					"heartbeat is in multicast mode");
			return;
		}
	}
	else if (strcmp(name, "mcast") != 0) {
		as_info_respond_error(db, AS_ERR_PARAMETER,
				"heartbeat is in mesh mode");
		return;
	}

	cf_dyn_buf_append_string(db, addr_port_str);
}

static void
cmd_health_outliers(const char* name, const char* params, cf_dyn_buf* db)
{
	as_health_get_outliers(db);
}

static void
cmd_health_stats(const char* name, const char* params, cf_dyn_buf* db)
{
	as_health_get_stats(db);
}

static void
cmd_help(const char* name, const char* params, cf_dyn_buf* db)
{
	bool is_ce = as_error_enterprise_only();

	for (uint32_t i = 0; i < N_SPECS; i++) {
		const as_info_cmd* cmd = &SPECS[i];

		// Filter client only commands and EE only commands when build is CE.
		if (cmd->client_only || (is_ce && cmd->ee_only)) {
			continue;
		}

		cf_dyn_buf_append_string(db, cmd->name);
		cf_dyn_buf_append_char(db, ';');
	}

	cf_dyn_buf_chomp_char(db, ';');
}

static void
cmd_histogram(const char* name, const char* params, cf_dyn_buf* db)
{
	char value_str[128];
	int  value_str_len = sizeof(value_str);
	info_param_result rv = as_info_param_get_namespace(params, value_str,
			&value_str_len);
	as_namespace* ns = NULL;

	if (! info_param_required_local_namespace_is_ok(db, value_str, &ns, rv)) {
		return;
	}

	value_str_len = sizeof(value_str);
	rv = as_info_parameter_get(params, "type", value_str, &value_str_len);

	if (! as_info_required_param_is_ok(db, "type", value_str, rv)) {
		return;
	}

	// Get optional set field - empty string is used.
	char set_name_str[AS_SET_NAME_MAX_SIZE] = { 0 };
	int set_name_str_len = sizeof(set_name_str);

	rv = as_info_parameter_get(params, "set", set_name_str, &set_name_str_len);
	rv = as_info_optional_param_is_ok(db, "set", set_name_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	as_namespace_get_hist_info(ns, set_name_str, value_str, db);
}

static void
cmd_index_pressure(const char* name, const char* params, cf_dyn_buf* db)
{
	for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
		as_namespace* ns = g_config.namespaces[i];
		cf_page_cache_stats stats;

		if (! cf_page_cache_get_stats(ns->arena, &stats)) {
			continue;
		}

		cf_dyn_buf_append_string(db, ns->name);
		cf_dyn_buf_append_char(db, ':');
		cf_dyn_buf_append_uint64(db, stats.resident);
		cf_dyn_buf_append_char(db, ':');
		cf_dyn_buf_append_uint64(db, stats.dirty);
		cf_dyn_buf_append_char(db, ';');
	}

	cf_dyn_buf_chomp(db);
}

static void
cmd_jem_stats(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// jem-stats:{file=<string>;options=<string>;sites=<string>}
	// where the "file", "options", and "sites" arguments are optional.
	//
	// Logs the JEMalloc statistics to the console or an optionally-specified file pathname.
	// Options may be a string containing any of the characters "gmablh", as defined by jemalloc(3) man page.
	// The "sites" parameter optionally specifies a file to dump memory accounting information to.
	// [Note:  Any options are only used if an output file is specified.]

	cf_debug(AS_INFO, "jem_stats command received: params %s", params);

	char file_str[100] = { 0 };
	int file_len = sizeof(file_str);
	info_param_result rv = as_info_parameter_get(params, "file", file_str,
			&file_len);

	rv = as_info_optional_param_is_ok(db, "file", file_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	char options_str[100] = { 0 };
	int options_len = sizeof(options_str);

	rv = as_info_parameter_get(params, "options", options_str, &options_len);
	rv = as_info_optional_param_is_ok(db, "options", options_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	char sites_str[100] = { 0 };
	int sites_len = sizeof(options_str);

	rv = as_info_parameter_get(params, "sites", sites_str, &sites_len);
	rv = as_info_optional_param_is_ok(db, "sites", sites_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	cf_alloc_log_stats(*file_str != '\0' ? file_str : NULL,
			*options_str != '\0' ? options_str : NULL);

	if (*sites_str != '\0') {
		cf_alloc_log_site_infos(sites_str);
	}

	as_info_respond_ok(db);
}

static void
cmd_latencies(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: latencies:[hist=<name>]
	// Example:
	// latencies:hist={test}-reads
	// output:
	// {test}-reads:msec,30618.2,0.05,0.01,0.00,0.00,0.00,0.00,0.00,0.00,0.00,0.00,0.00,0.00,0.00,0.00,0.00,0.00,0.00
	// Explanation:
	// <name>,units,TPS, ...
	// Values following the TPS are percentages exceeding logarithmic thresholds.

	cf_debug(AS_INFO, "%s command received: params %s", name, params);

	char value_str[100];
	int  value_str_len = sizeof(value_str);
	info_param_result rv = as_info_parameter_get(params, "hist", value_str,
			&value_str_len);

	rv = as_info_optional_param_is_ok(db, "hist", value_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK_NOT_FOUND) {
		// Canonical histograms.

		histogram_get_latencies(g_stats.batch_index_hist, db);

		for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
			as_namespace* ns = g_config.namespaces[i];

			histogram_get_latencies(ns->read_hist, db);
			histogram_get_latencies(ns->write_hist, db);
			histogram_get_latencies(ns->udf_hist, db);
			histogram_get_latencies(ns->batch_sub_read_hist, db);
			histogram_get_latencies(ns->batch_sub_write_hist, db);
			histogram_get_latencies(ns->batch_sub_udf_hist, db);
			histogram_get_latencies(ns->pi_query_hist, db);
			histogram_get_latencies(ns->si_query_hist, db);
		}
	}
	else {
		// Named histograms.

		if (strcmp(value_str, "batch-index") == 0) {
			histogram_get_latencies(g_stats.batch_index_hist, db);
		}
		else if (strcmp(value_str, "info") == 0) {
			histogram_get_latencies(g_stats.info_hist, db);
		}
		else if (strcmp(value_str, "benchmarks-fabric") == 0) {
			histogram_get_latencies(g_stats.fabric_send_init_hists[AS_FABRIC_CHANNEL_BULK], db);
			histogram_get_latencies(g_stats.fabric_send_fragment_hists[AS_FABRIC_CHANNEL_BULK], db);
			histogram_get_latencies(g_stats.fabric_recv_fragment_hists[AS_FABRIC_CHANNEL_BULK], db);
			histogram_get_latencies(g_stats.fabric_recv_cb_hists[AS_FABRIC_CHANNEL_BULK], db);
			histogram_get_latencies(g_stats.fabric_send_init_hists[AS_FABRIC_CHANNEL_CTRL], db);
			histogram_get_latencies(g_stats.fabric_send_fragment_hists[AS_FABRIC_CHANNEL_CTRL], db);
			histogram_get_latencies(g_stats.fabric_recv_fragment_hists[AS_FABRIC_CHANNEL_CTRL], db);
			histogram_get_latencies(g_stats.fabric_recv_cb_hists[AS_FABRIC_CHANNEL_CTRL], db);
			histogram_get_latencies(g_stats.fabric_send_init_hists[AS_FABRIC_CHANNEL_META], db);
			histogram_get_latencies(g_stats.fabric_send_fragment_hists[AS_FABRIC_CHANNEL_META], db);
			histogram_get_latencies(g_stats.fabric_recv_fragment_hists[AS_FABRIC_CHANNEL_META], db);
			histogram_get_latencies(g_stats.fabric_recv_cb_hists[AS_FABRIC_CHANNEL_META], db);
			histogram_get_latencies(g_stats.fabric_send_init_hists[AS_FABRIC_CHANNEL_RW], db);
			histogram_get_latencies(g_stats.fabric_send_fragment_hists[AS_FABRIC_CHANNEL_RW], db);
			histogram_get_latencies(g_stats.fabric_recv_fragment_hists[AS_FABRIC_CHANNEL_RW], db);
			histogram_get_latencies(g_stats.fabric_recv_cb_hists[AS_FABRIC_CHANNEL_RW], db);
		}
		else if (*value_str == '{') {
			// Named namespace-scoped histogram - parse '{namespace}-' prefix.

			char* ns_name = value_str + 1;
			char* ns_name_end = strchr(ns_name, '}');
			as_namespace* ns = as_namespace_get_bybuf((uint8_t*)ns_name, ns_name_end - ns_name);

			if (ns == NULL) {
				cf_info(AS_INFO, "%s command: unrecognized histogram: %s", name, value_str);
				as_info_respond_error(db, AS_ERR_PARAMETER, "bad hist name");
				return;
			}

			char* hist_name = ns_name_end + 1;

			if (*hist_name++ != '-') {
				cf_info(AS_INFO, "%s command: unrecognized histogram: %s", name, value_str);
				as_info_respond_error(db, AS_ERR_PARAMETER, "bad hist name");
				return;
			}

			if (strcmp(hist_name, "read") == 0) {
				histogram_get_latencies(ns->read_hist, db);
			}
			else if (strcmp(hist_name, "write") == 0) {
				histogram_get_latencies(ns->write_hist, db);
			}
			else if (strcmp(hist_name, "udf") == 0) {
				histogram_get_latencies(ns->udf_hist, db);
			}
			else if (strcmp(hist_name, "batch-sub-read") == 0) {
				histogram_get_latencies(ns->batch_sub_read_hist, db);
			}
			else if (strcmp(hist_name, "batch-sub-write") == 0) {
				histogram_get_latencies(ns->batch_sub_write_hist, db);
			}
			else if (strcmp(hist_name, "batch-sub-udf") == 0) {
				histogram_get_latencies(ns->batch_sub_udf_hist, db);
			}
			else if (strcmp(hist_name, "pi-query") == 0) {
				histogram_get_latencies(ns->pi_query_hist, db);
			}
			else if (strcmp(hist_name, "si-query") == 0) {
				histogram_get_latencies(ns->si_query_hist, db);
			}
			else if (strcmp(hist_name, "read-touch") == 0) {
				histogram_get_latencies(ns->read_touch_hist, db);
			}
			else if (strcmp(hist_name, "re-repl") == 0) {
				histogram_get_latencies(ns->re_repl_hist, db);
			}
			else if (strcmp(hist_name, "proxy") == 0) {
				histogram_get_latencies(ns->proxy_hist, db);
			}
			else if (strcmp(hist_name, "benchmarks-read") == 0) {
				histogram_get_latencies(ns->read_start_hist, db);
				histogram_get_latencies(ns->read_restart_hist, db);
				histogram_get_latencies(ns->read_dup_res_hist, db);
				histogram_get_latencies(ns->read_repl_ping_hist, db);
				histogram_get_latencies(ns->read_local_hist, db);
				histogram_get_latencies(ns->read_response_hist, db);
			}
			else if (strcmp(hist_name, "benchmarks-write") == 0) {
				histogram_get_latencies(ns->write_start_hist, db);
				histogram_get_latencies(ns->write_restart_hist, db);
				histogram_get_latencies(ns->write_dup_res_hist, db);
				histogram_get_latencies(ns->write_master_hist, db);
				histogram_get_latencies(ns->write_repl_write_hist, db);
				histogram_get_latencies(ns->write_response_hist, db);
			}
			else if (strcmp(hist_name, "benchmarks-udf") == 0) {
				histogram_get_latencies(ns->udf_start_hist, db);
				histogram_get_latencies(ns->udf_restart_hist, db);
				histogram_get_latencies(ns->udf_dup_res_hist, db);
				histogram_get_latencies(ns->udf_master_hist, db);
				histogram_get_latencies(ns->udf_repl_write_hist, db);
				histogram_get_latencies(ns->udf_response_hist, db);
			}
			else if (strcmp(hist_name, "benchmarks-batch-sub") == 0) {
				histogram_get_latencies(ns->batch_sub_prestart_hist, db);
				histogram_get_latencies(ns->batch_sub_start_hist, db);
				histogram_get_latencies(ns->batch_sub_restart_hist, db);
				histogram_get_latencies(ns->batch_sub_dup_res_hist, db);
				histogram_get_latencies(ns->batch_sub_repl_ping_hist, db);
				histogram_get_latencies(ns->batch_sub_read_local_hist, db);
				histogram_get_latencies(ns->batch_sub_write_master_hist, db);
				histogram_get_latencies(ns->batch_sub_udf_master_hist, db);
				histogram_get_latencies(ns->batch_sub_repl_write_hist, db);
				histogram_get_latencies(ns->batch_sub_response_hist, db);
			}
			else if (strcmp(hist_name, "benchmarks-udf-sub") == 0) {
				histogram_get_latencies(ns->udf_sub_start_hist, db);
				histogram_get_latencies(ns->udf_sub_restart_hist, db);
				histogram_get_latencies(ns->udf_sub_dup_res_hist, db);
				histogram_get_latencies(ns->udf_sub_master_hist, db);
				histogram_get_latencies(ns->udf_sub_repl_write_hist, db);
				histogram_get_latencies(ns->udf_sub_response_hist, db);
			}
			else if (strcmp(hist_name, "benchmarks-ops-sub") == 0) {
				histogram_get_latencies(ns->ops_sub_start_hist, db);
				histogram_get_latencies(ns->ops_sub_restart_hist, db);
				histogram_get_latencies(ns->ops_sub_dup_res_hist, db);
				histogram_get_latencies(ns->ops_sub_master_hist, db);
				histogram_get_latencies(ns->ops_sub_repl_write_hist, db);
				histogram_get_latencies(ns->ops_sub_response_hist, db);
			}
			else {
				cf_info(AS_INFO, "%s command: unrecognized histogram: %s", name, value_str);
				as_info_respond_error(db, AS_ERR_PARAMETER, "bad hist name");
				return;
			}
		}
		else {
			cf_info(AS_INFO, "%s command: unrecognized histogram: %s", name, value_str);
			as_info_respond_error(db, AS_ERR_PARAMETER, "bad hist name");
			return;
		}
	}

	cf_dyn_buf_chomp(db);
}

static void
cmd_log(const char* name, const char* params, cf_dyn_buf* db)
{
	int sink_id;
	char* context = strchr(params, ARG_SEP);

	if (context != NULL) { // this means: log/id/context,
		*context = 0;
		context++;

		if (cf_str_atoi(params, &sink_id) != 0) {
			as_info_respond_error(db, AS_ERR_PARAMETER, "bad sink-id");
			return;
		}

		cf_log_get_level(sink_id, context, db);
	}
	else { // this means just: log/id , so get all contexts
		if (cf_str_atoi(params, &sink_id) != 0) {
			as_info_respond_error(db, AS_ERR_PARAMETER, "bad sink-id");
			return;
		}

		cf_log_get_all_levels(sink_id, db);
	}
}

static void
cmd_log_message(const char* name, const char* params, cf_dyn_buf* db)
{
	// Log a message to the server. Limited to 2048 characters.
	// Command format: log-message:message=<MESSAGE>[;who=<WHO>]
	// Example: log-message:message=Example Log Message;who=Aerospike User

	char who[128];
	int who_len = sizeof(who);
	info_param_result rv = as_info_parameter_get(params, "who", who, &who_len);

	rv = as_info_optional_param_is_ok(db, "who", who, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK_NOT_FOUND) {
		strcpy(who, "unknown");
	}

	char message[2048];
	int message_len = sizeof(message);

	rv = as_info_parameter_get(params, "message", message, &message_len);

	if (! as_info_required_param_is_ok(db, "message", message, rv)) {
		return;
	}

	cf_info(AS_INFO, "%s: %s", who, message);
}

static void
cmd_log_set(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: log-set:id=<id>;<context>=<level>
	// Example: log-set:id=0;service=detail

	cf_debug(AS_INFO, "received log-set:%s", params);

	char* params_cpy = strdup(params);

	char* save_ptr = NULL;
	const char* tok = strtok_r(params_cpy, "=", &save_ptr);

	if (tok == NULL || strcmp(tok, "id") != 0) {
		cf_warning(AS_INFO, "log-set: missing id");
		as_info_respond_error(db, AS_ERR_PARAMETER, "missing id");
		cf_free(params_cpy);
		return;
	}

	const char* id_str = strtok_r(NULL, ";", &save_ptr);
	uint32_t id;

	if (id_str == NULL || cf_strtoul_u32(id_str, &id) != 0) {
		cf_warning(AS_INFO, "log-set: bad id");
		as_info_respond_error(db, AS_ERR_PARAMETER, "bad id");
		cf_free(params_cpy);
		return;
	}

	const char* context_str = strtok_r(NULL, "=", &save_ptr);

	if (context_str == NULL) {
		cf_warning(AS_INFO, "log-set: missing context");
		as_info_respond_error(db, AS_ERR_PARAMETER, "missing context");
		cf_free(params_cpy);
		return;
	}

	const char* level_str = strtok_r(NULL, ";", &save_ptr);

	if (level_str == NULL) {
		cf_warning(AS_INFO, "log-set: bad level");
		as_info_respond_error(db, AS_ERR_PARAMETER, "bad level");
		cf_free(params_cpy);
		return;
	}

	if (! cf_log_set_level(id, context_str, level_str)) {
		as_info_respond_error(db, AS_ERR_PARAMETER, "bad parameter");
		cf_free(params_cpy);
		return;
	}

	cf_info(AS_INFO, "log-set:id=%s:%s=%s", id_str, context_str, level_str);
	as_info_respond_ok(db);
	cf_free(params_cpy);
}

static void
cmd_logs(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_log_get_sinks(db);
}

static void
cmd_namespace(const char* name, const char* params, cf_dyn_buf* db)
{
	as_namespace* ns = as_namespace_get_byname(params);

	if (ns == NULL) {
		// For now, no info_respond_error(), clients won't handle error.
		cf_dyn_buf_append_string(db, "type=unknown");
		return;
	}

	info_get_namespace_info(ns, db);
	as_cfg_info_namespace_config_get(ns, db);

	cf_dyn_buf_chomp(db);
}

static void
cmd_namespaces(const char* name, const char* params, cf_dyn_buf* db)
{
	for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
		cf_dyn_buf_append_string(db, g_config.namespaces[i]->name);
		cf_dyn_buf_append_char(db, ';');
	}

	if (g_config.n_namespaces > 0) {
		cf_dyn_buf_chomp(db);
	}
}

static void
cmd_node(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_uint64_x(db, g_config.self_node);
}

static void
cmd_objects(const char* name, const char* params, cf_dyn_buf* db)
{
	uint64_t objects = 0;

	for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
		objects += g_config.namespaces[i]->n_objects;
	}

	cf_dyn_buf_append_uint64(db, objects);
}

static void
cmd_partition_generation(const char* name, const char* params, cf_dyn_buf* db)
{
	// TODO - ARM TSO plugin - will need acquire semantic.
	cf_dyn_buf_append_int(db, g_partition_generation);
}

static void
cmd_partition_info(const char* name, const char* params, cf_dyn_buf* db)
{
	as_partition_getinfo_str(db);
}

static void
cmd_partitions(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_uint32(db, AS_PARTITIONS);
}

static void
cmd_physical_devices(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: "physical-devices:path=<path>"
	//
	// <path> can specify a device partition, file path, mount directory, etc.
	// ... anything backed by one or more physical devices.

	char path_str[1024];
	int path_str_len = (int)sizeof(path_str);
	info_param_result rv = as_info_parameter_get(params, "path", path_str,
			&path_str_len);

	if (! as_info_required_param_is_ok(db, "path", path_str, rv)) {
		return;
	}

	cf_storage_device_info* device_info = cf_storage_get_device_info(path_str);

	if (device_info == NULL) {
		cf_warning(AS_INFO, "can't get device info for %s", path_str);
		as_info_respond_error(db, AS_ERR_NOT_FOUND, "no device info");
		return;
	}

	for (uint32_t i = 0; i < device_info->n_phys; i++) {
		cf_dyn_buf_append_string(db, "physical-device=");
		cf_dyn_buf_append_string(db, device_info->phys[i].dev_path);
		cf_dyn_buf_append_char(db, ':');
		cf_dyn_buf_append_string(db, "age=");
		cf_dyn_buf_append_int(db, device_info->phys[i].nvme_age);

		cf_dyn_buf_append_char(db, ';');
	}

	cf_dyn_buf_chomp(db);
}

static void
cmd_query_abort_all(const char* name, const char* params, cf_dyn_buf* db)
{
	uint32_t n_queries_killed = as_query_manager_abort_all_jobs();

	cf_dyn_buf_append_string(db, "number-of-queries-killed:");
	cf_dyn_buf_append_uint32(db, n_queries_killed);
}

static void
cmd_query_abort(const char* name, const char* params, cf_dyn_buf* db)
{
	(void)name;

	char trid_str[1 + 24 + 1]; // allow octal, decimal, hex
	int trid_str_len = (int)sizeof(trid_str);
	// Allow 'id' for backward compatibility of scan-abort. Remove in 6 months.
	const char* aliases[] = { "trid", "id", NULL };
	info_param_result rv = as_info_param_get_aliases(params, aliases, trid_str,
			&trid_str_len);

	if (! as_info_required_param_is_ok(db, "trid", trid_str, rv)) {
		return;
	}

	uint64_t trid = 0;

	if (cf_strtoul_u64_raw(trid_str, &trid) != 0 || trid == 0) {
		cf_warning(AS_INFO, "bad trid %s", trid_str);
		as_info_respond_error(db, AS_ERR_PARAMETER, "bad trid");
		return;
	}

	if (as_query_manager_abort_job(trid)) {
		as_info_respond_ok(db);
		return;
	}

	as_info_respond_error(db, AS_ERR_NOT_FOUND, "trid not active");
}

static void
cmd_query_show(const char* name, const char* params, cf_dyn_buf* db)
{
	(void)name;

	char trid_str[1 + 24 + 1]; // allow octal, decimal, hex
	int trid_str_len = (int)sizeof(trid_str);

	info_param_result rv = as_info_parameter_get(params, "trid", trid_str,
			&trid_str_len);

	rv = as_info_optional_param_is_ok(db, "trid", trid_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK_NOT_FOUND) { // no trid specified - show all
		as_query_manager_get_all_jobs_info(db);
		return;
	}

	uint64_t trid = 0;

	if (cf_strtoul_u64_raw(trid_str, &trid) != 0 || trid == 0) {
		cf_warning(AS_INFO, "bad trid %s", trid_str);
		as_info_respond_error(db, AS_ERR_PARAMETER, "bad trid");
		return;
	}

	as_query_manager_get_job_info(trid, db);
}

static void
cmd_quiesce(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: quiesce:

	if (g_config.stay_quiesced) {
		as_info_respond_error(db, AS_ERR_FORBIDDEN,
				"permanently quiesced");
		return;
	}

	for (uint32_t ns_ix = 0; ns_ix < g_config.n_namespaces; ns_ix++) {
		g_config.namespaces[ns_ix]->pending_quiesce = true;
	}

	as_info_respond_ok(db);

	cf_info(AS_INFO, "quiesced this node");
}

static void
cmd_quiesce_undo(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: quiesce-undo:

	if (g_config.stay_quiesced) {
		cf_dyn_buf_append_string(db, "ignored-permanently-quiesced");
		return;
	}

	for (uint32_t ns_ix = 0; ns_ix < g_config.n_namespaces; ns_ix++) {
		g_config.namespaces[ns_ix]->pending_quiesce = false;
	}

	as_info_respond_ok(db);

	cf_info(AS_INFO, "un-quiesced this node");
}

static void
cmd_rack_ids(const char* name, const char* params, cf_dyn_buf* db)
{
	as_partition_balance_effective_rack_ids(db);
}

static void
cmd_racks(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: racks:{namespace=<namespace-name>}

	char param_str[AS_ID_NAMESPACE_SZ] = { 0 };
	int param_str_len = (int)sizeof(param_str);
	info_param_result rv = as_info_param_get_namespace(params, param_str,
			&param_str_len);
	as_namespace* ns = NULL;

	rv = info_param_optional_local_namespace_is_ok(db, param_str, &ns, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		as_exchange_info_lock();

		namespace_rack_info(ns, db, ns->rack_ids, ns->cluster_size,
				ns->succession, "rack_");

		if (ns->roster_count != 0) {
			cf_dyn_buf_append_char(db, ':');
			namespace_rack_info(ns, db, ns->roster_rack_ids, ns->roster_count,
					ns->roster, "roster_rack_");
		}

		as_exchange_info_unlock();

		return;
	}

	for (uint32_t ns_ix = 0; ns_ix < g_config.n_namespaces; ns_ix++) {
		ns = g_config.namespaces[ns_ix];

		cf_dyn_buf_append_string(db, "ns=");
		cf_dyn_buf_append_string(db, ns->name);
		cf_dyn_buf_append_char(db, ':');

		as_exchange_info_lock();

		namespace_rack_info(ns, db, ns->rack_ids, ns->cluster_size,
				ns->succession, "rack_");

		if (ns->roster_count != 0) {
			cf_dyn_buf_append_char(db, ':');
			namespace_rack_info(ns, db, ns->roster_rack_ids, ns->roster_count,
					ns->roster, "roster_rack_");
		}

		as_exchange_info_unlock();

		cf_dyn_buf_append_char(db, ';');
	}

	cf_dyn_buf_chomp(db);
}

static void
cmd_rebalance_generation(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_uint64(db, g_rebalance_generation);
}

static void
cmd_recluster(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: recluster:

	int rv = as_clustering_cluster_reform();

	// TODO - resolve error condition further?
	if (rv == 0) {
		as_info_respond_ok(db);
		return;
	}

	if (rv == 1) {
		cf_dyn_buf_append_string(db, "ignored-by-non-principal");
		return;
	}

	as_info_respond_error(db, AS_ERR_UNKNOWN, "unknown");
}

static void
cmd_replicas(const char* name, const char* params, cf_dyn_buf* db)
{
	char max_str[4];
	int len = (int)sizeof(max_str);
	info_param_result rv = as_info_parameter_get(params, "max", max_str, &len);

	rv = as_info_optional_param_is_ok(db, "max", max_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	uint32_t max_repls = 0;

	if (rv == INFO_PARAM_OK && cf_str_atoi_u32(max_str, &max_repls) != 0) {
		cf_warning(AS_INFO, "non-integer max parameter");
		as_info_respond_error(db, AS_ERR_PARAMETER, "bad max");
		return;
	}

	as_partition_get_replicas_all_str(db, true, max_repls);
}

static void
cmd_replicas_all(const char* name, const char* params, cf_dyn_buf* db)
{
	as_partition_get_replicas_all_str(db, false, 0);
}

static void
cmd_replicas_master(const char* name, const char* params, cf_dyn_buf* db)
{
	as_partition_get_replicas_master_str(db);
}

static void
cmd_revive(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: revive:{namespace=<ns-name>}

	char ns_name[AS_ID_NAMESPACE_SZ] = { 0 };
	int ns_name_len = (int)sizeof(ns_name);
	info_param_result rv = as_info_param_get_namespace(params, ns_name,
			&ns_name_len);
	as_namespace* ns = NULL;

	rv = info_param_optional_local_namespace_is_ok(db, ns_name, &ns, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		if (! as_partition_balance_revive(ns)) {
			cf_warning(AS_INFO, "revive: failed - recluster in progress");
			as_info_respond_error(db, AS_ERR_CLUSTER_KEY_MISMATCH,
					"failed revive");
			return;
		}

		cf_info(AS_INFO, "revive: complete - issue 'recluster:' command");
		as_info_respond_ok(db);
		return;
	}

	for (uint32_t ns_ix = 0; ns_ix < g_config.n_namespaces; ns_ix++) {
		as_namespace* ns = g_config.namespaces[ns_ix];

		if (! as_partition_balance_revive(ns)) {
			cf_warning(AS_INFO, "revive: failed - recluster in progress");
			as_info_respond_error(db, AS_ERR_CLUSTER_KEY_MISMATCH,
					"failed revive");
			return;
		}
	}

	cf_info(AS_INFO, "revive: complete - issue 'recluster:' command");
	as_info_respond_ok(db);
}

static void
cmd_roster(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: roster:{namespace=<ns-name>}

	char ns_name[AS_ID_NAMESPACE_SZ] = { 0 };
	int ns_name_len = (int)sizeof(ns_name);
	info_param_result rv = as_info_param_get_namespace(params, ns_name,
			&ns_name_len);
	as_namespace* ns = NULL;

	rv = info_param_optional_local_namespace_is_ok(db, ns_name, &ns, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		namespace_roster_info(ns, db);
		return;
	}

	for (uint32_t ns_ix = 0; ns_ix < g_config.n_namespaces; ns_ix++) {
		as_namespace* ns = g_config.namespaces[ns_ix];

		cf_dyn_buf_append_string(db, "ns=");
		cf_dyn_buf_append_string(db, ns->name);
		cf_dyn_buf_append_char(db, ':');

		namespace_roster_info(ns, db);

		cf_dyn_buf_append_char(db, ';');
	}

	cf_dyn_buf_chomp(db);
}

static void
cmd_roster_set(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// roster-set:namespace=<ns-name>;nodes=<nodes-string>
	// where <nodes-string> is comma-separated list of node-id:rack-id pairs, and
	// the :rack-id may be absent, indicating a rack-id of 0.

	// Get the namespace name.

	char ns_name[AS_ID_NAMESPACE_SZ];
	int ns_name_len = (int)sizeof(ns_name);
	int ns_rv = as_info_param_get_namespace(params, ns_name, &ns_name_len);

	if (! as_info_required_param_is_ok(db, "namespace", ns_name, ns_rv)) {
		return;
	}

	// Get the nodes list.

	char nodes[AS_CLUSTER_SZ*  ROSTER_STRING_ELE_LEN];
	int nodes_len = (int)sizeof(nodes);
	int nodes_rv = as_info_parameter_get(params, "nodes", nodes, &nodes_len);

	if (! as_info_required_param_is_ok(db, "nodes", nodes, nodes_rv)) {
		return;
	}

	// Issue the roster-set command.

	as_roster_set_nodes_cmd(ns_name, nodes, db);
}

static void
cmd_sets(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// format w/o namespace is
	//   ns1:set1:prop1=val1:prop2=val2:..propn=valn;ns1:set2...;ns2:set1...;
	// format w namespace w/o set name is
	//   ns:set1:prop1=val1:prop2=val2...propn=valn;ns:set2...;
	// format w namespace & set name is
	//   prop1=val1:prop2=val2...propn=valn;

	char* set_name = NULL;
	as_namespace* ns = NULL;

	if (strlen(params) > 0) {
		set_name = strchr(params, ARG_SEP);

		if (set_name != NULL) {
			int ns_name_len = (set_name - params);
			char ns_name[ns_name_len + 1];

			memcpy(ns_name, params, ns_name_len);
			ns_name[ns_name_len] = '\0';
			ns = as_namespace_get_byname(ns_name);
			set_name++;
		}
		else {
			ns = as_namespace_get_byname(params);
		}

		if (ns == NULL) {
			// For now, no info_respond_error(), clients won't handle error.
			cf_dyn_buf_append_string(db, "ns_type=unknown");
			return;
		}
	}

	if (ns == NULL) {
		for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
			as_namespace_get_set_info(g_config.namespaces[i], set_name, db);
		}
	}
	else {
		as_namespace_get_set_info(ns, set_name, db);
	}
}

static void
cmd_sindex(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// format w/o namespace is:
	//    ns=ns1:set=set1:indexname=index1:prop1=val1:...:propn=valn;ns=ns1:set=set2:indexname=index2:...;ns=ns2:set=set1:...;
	// format w namespace w/o index name is:
	//    ns=ns1:set=set1:indexname=index1:prop1=val1:...:propn=valn;ns=ns1:set=set2:indexname=indexname2:...;
	// format w namespace & index name is:
	//    prop1=val1;prop2=val2;...;propn=valn

	char* index_name = NULL;
	as_namespace* ns = NULL;

	if (strlen(params) > 0) {
		index_name = strchr(params, ARG_SEP);

		// pull out namespace, and namespace name...
		if (index_name != NULL) {
			int ns_name_len = (index_name - params);
			char ns_name[ns_name_len + 1];

			memcpy(ns_name, params, ns_name_len);
			ns_name[ns_name_len] = '\0';
			ns = as_namespace_get_byname(ns_name);
			index_name++; // currently points to the ARG_SEP, which is not what we want.
		}
		else {
			ns = as_namespace_get_byname(params);
		}

		if (ns == NULL) {
			// For now, no info_respond_error(), clients won't handle error.
			cf_dyn_buf_append_string(db, "ns_type=unknown");
			return;
		}
	}

	if (ns == NULL) {
		for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
			as_sindex_list_str(g_config.namespaces[i], false, db);
		}

		cf_dyn_buf_chomp_char(db, ';');
	}
	else if (index_name == NULL) {
		as_sindex_list_str(ns, false, db);
		cf_dyn_buf_chomp_char(db, ';');
	}
	else if (! as_sindex_stats_str(ns, index_name, db)) {
		cf_warning(AS_INFO, "failed to get statistics for index %s: not found",
				index_name);
		as_info_respond_error(db, AS_ERR_SINDEX_NOT_FOUND, "no index");
	}
}

static void
cmd_sindex_create(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// sindex-create:ns=usermap;set=demo;indexname=um_age;indextype=list;indexdata=age,numeric
	// sindex-create:ns=usermap;set=demo;indexname=um_state;indexdata=state,string
	// sindex-create:ns=usermap;set=demo;indexname=um_highscore;context=<base64-cdt-ctx>;indexdata=scores,numeric

	char index_name_str[INAME_MAX_SZ];
	int index_name_len = sizeof(index_name_str);
	info_param_result rv = as_info_parameter_get(params, "indexname",
			index_name_str, &index_name_len);

	if (! as_info_required_param_is_ok(db, "indexname", index_name_str, rv)) {
		return;
	}

	char ns_str[AS_ID_NAMESPACE_SZ];
	int ns_len = sizeof(ns_str);

	rv = as_info_param_get_namespace_ns(params, ns_str, &ns_len);

	if (! as_info_required_param_is_ok(db, "namespace", ns_str, rv)) {
		return;
	}

	char set_str[AS_SET_NAME_MAX_SIZE];
	char* p_set_str = NULL;
	int set_len = sizeof(set_str);

	rv = as_info_parameter_get(params, "set", set_str, &set_len);
	rv = as_info_optional_param_is_ok(db, "set", set_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		p_set_str = set_str;
	}

	char ctx_b64[CTX_B64_MAX_SZ];
	int ctx_b64_len = sizeof(ctx_b64);
	const char* p_cdt_ctx = NULL;

	rv = as_info_parameter_get(params, "context", ctx_b64, &ctx_b64_len);
	rv = as_info_optional_param_is_ok(db, "context", ctx_b64, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		uint8_t* buf;
		int32_t buf_sz = as_sindex_cdt_ctx_b64_decode(ctx_b64, ctx_b64_len,
				&buf);

		if (buf_sz > 0) {
			cf_free(buf);
		}
		else if (buf_sz < 0) {
			switch (buf_sz) {
			case -1:
				cf_warning(AS_INFO, "sindex-create %s: 'context' invalid base64",
						index_name_str);
				as_info_respond_error(db, AS_ERR_PARAMETER,
						"'context' invalid base64");
				return;
			case -2:
				cf_warning(AS_INFO, "sindex-create %s: 'context' invalid cdt context",
						index_name_str);
				as_info_respond_error(db, AS_ERR_PARAMETER,
						"'context' invalid cdt context");
				return;
			case -3:
				cf_warning(AS_INFO, "sindex-create %s: 'context' not normalized msgpack",
						index_name_str);
				as_info_respond_error(db, AS_ERR_PARAMETER,
						"'context' not normalized msgpack");
				return;
			default:
				cf_crash(AS_INFO, "unreachable");
			}
		}

		p_cdt_ctx = ctx_b64;
	}

	char indextype_str[INDEXTYPE_MAX_SZ];
	int indtype_len = sizeof(indextype_str);
	as_sindex_type itype;

	rv = as_info_parameter_get(params, "indextype", indextype_str,
			&indtype_len);
	rv = as_info_optional_param_is_ok(db, "indextype", indextype_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK_NOT_FOUND) {
		// If not specified, the index type is DEFAULT.
		itype = AS_SINDEX_ITYPE_DEFAULT;
	}
	else {
		itype = as_sindex_itype_from_string(indextype_str);

		if (itype == AS_SINDEX_N_ITYPES) {
			cf_warning(AS_INFO, "sindex-create %s: bad 'indextype' '%s'",
					index_name_str, indextype_str);
			as_info_respond_error(db, AS_ERR_PARAMETER,
					"bad 'indextype' - must be one of 'default', 'list', 'mapkeys', 'mapvalues'");
			return;
		}
	}

	// indexdata=bin-name,keytype
	char indexdata_str[INDEXDATA_MAX_SZ];
	int indexdata_len = sizeof(indexdata_str);

	rv = as_info_parameter_get(params, "indexdata", indexdata_str,
			&indexdata_len);

	if (! as_info_required_param_is_ok(db, "indexdata", indexdata_str, rv)) {
		return;
	}

	char* bin_name = indexdata_str;
	char* type_str = strchr(indexdata_str, ',');

	if (type_str == NULL) {
		cf_warning(AS_INFO, "sindex-create %s: 'indexdata' missing bin type",
				index_name_str);
		as_info_respond_error(db, AS_ERR_PARAMETER,
				"'indexdata' missing bin type");
		return;
	}

	*type_str++ = '\0';

	if (bin_name[0] == '\0') {
		cf_warning(AS_INFO, "sindex-create %s: 'indexdata' missing bin name",
				index_name_str);
		as_info_respond_error(db, AS_ERR_PARAMETER,
				"'indexdata' missing bin name");
		return;
	}

	if (strlen(bin_name) >= AS_BIN_NAME_MAX_SZ) {
		cf_warning(AS_INFO, "sindex-create %s: 'indexdata' bin name too long",
				index_name_str);
		as_info_respond_error(db, AS_ERR_PARAMETER,
				"'indexdata' bin name too long");
		return;
	}

	as_particle_type ktype = as_sindex_ktype_from_string(type_str);

	if (ktype == AS_PARTICLE_TYPE_BAD) {
		cf_warning(AS_INFO, "sindex-create %s: bad 'indexdata' bin type '%s'",
				index_name_str, type_str);
		as_info_respond_error(db, AS_ERR_PARAMETER,
				"bad 'indexdata' bin type - must be one of 'numeric', 'string', 'blob', 'geo2dsphere'");
		return;
	}

	if (itype == AS_SINDEX_ITYPE_MAPKEYS &&
			ktype != AS_PARTICLE_TYPE_INTEGER &&
			ktype != AS_PARTICLE_TYPE_STRING &&
			ktype != AS_PARTICLE_TYPE_BLOB) {
		cf_warning(AS_INFO, "sindex-create %s: bad 'indexdata' bin type '%s' for 'indextype' 'mapkeys'",
				index_name_str, type_str);
		as_info_respond_error(db, AS_ERR_PARAMETER,
				"bad 'indexdata' bin type for 'indextype' 'mapkeys' - must be one of 'numeric', 'string', 'blob'");
		return;
	}

	cf_info(AS_INFO, "sindex-create: request received for %s:%s via info",
			ns_str, index_name_str);

	char smd_key[SINDEX_SMD_KEY_MAX_SZ];

	as_sindex_build_smd_key(ns_str, p_set_str, bin_name, p_cdt_ctx, itype,
			ktype, smd_key);

	find_sindex_key_udata fsk = {
			.ns_name = ns_str,
			.index_name = index_name_str,
			.smd_key = smd_key
	};

	as_smd_get_all(AS_SMD_MODULE_SINDEX, find_sindex_key, &fsk);

	if (fsk.found_key != NULL) {
		if (strcmp(fsk.found_key, smd_key) != 0) {
			cf_free(fsk.found_key);
			cf_warning(AS_INFO, "sindex-create %s:%s: 'indexname' already exists with different definition",
					ns_str, index_name_str);
			as_info_respond_error(db, AS_ERR_SINDEX_FOUND,
					"'indexname' already exists with different definition");
			return;
		}

		cf_free(fsk.found_key);
		cf_info(AS_INFO, "sindex-create %s:%s: 'indexname' and defintion already exists",
				ns_str, index_name_str);
		as_info_respond_ok(db);
		return;
	}

	if (fsk.n_name_matches > 1) {
		cf_warning(AS_INFO, "sindex-create %s:%s: 'indexname' already exists with %u definitions - rename(s) required",
				ns_str, index_name_str, fsk.n_name_matches);
		as_info_respond_error(db, AS_ERR_SINDEX_FOUND,
				"'indexname' already exists with multiple definitions");
		return;
	}

	if (! fsk.has_smd_key && fsk.n_indexes >= MAX_N_SINDEXES) {
		cf_warning(AS_INFO, "sindex-create %s:%s: already at sindex definition limit",
				ns_str, index_name_str);
		as_info_respond_error(db, AS_ERR_SINDEX_MAX_COUNT,
				"already at sindex definition limit");
		return;
	}

	if (! as_smd_set_blocking(AS_SMD_MODULE_SINDEX, smd_key, index_name_str,
			0)) {
		cf_warning(AS_INFO, "sindex-create: timeout while creating %s:%s in SMD",
				ns_str, index_name_str);
		as_info_respond_error(db, AS_ERR_TIMEOUT, "timeout");
		return;
	}

	as_info_respond_ok(db);
}

static void
cmd_sindex_delete(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// sindex-delete:ns=usermap;set=demo;indexname=um_state

	char index_name_str[INAME_MAX_SZ];
	int index_name_len = sizeof(index_name_str);
	info_param_result rv = as_info_parameter_get(params, "indexname", index_name_str,
			&index_name_len);

	rv = as_info_required_param_is_ok(db, "indexname", index_name_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	char ns_str[AS_ID_NAMESPACE_SZ];
	int ns_len = sizeof(ns_str);

	rv = as_info_param_get_namespace_ns(params, ns_str, &ns_len);

	if (! as_info_required_param_is_ok(db, "namespace", ns_str, rv)) {
		return;
	}

	cf_info(AS_INFO, "sindex-delete: request received for %s:%s via info",
			ns_str, index_name_str);

	find_sindex_key_udata fsk = {
			.ns_name = ns_str,
			.index_name = index_name_str
	};

	as_smd_get_all(AS_SMD_MODULE_SINDEX, find_sindex_key, &fsk);

	if (fsk.found_key == NULL) {
		if (fsk.n_name_matches == 0) {
			cf_info(AS_INFO, "sindex-delete: 'indexname' %s not found",
					fsk.index_name);
			as_info_respond_ok(db);
			return;
		}

		cf_warning(AS_INFO, "sindex-delete: 'indexname' %s not unique - found %u matches - rename(s) required",
				fsk.index_name, fsk.n_name_matches);
		as_info_respond_error(db, AS_ERR_SINDEX_FOUND,
				"'indexname' is not unique");
		return;
	}

	if (! as_smd_delete_blocking(AS_SMD_MODULE_SINDEX, fsk.found_key, 0)) {
		cf_free(fsk.found_key);
		cf_warning(AS_INFO, "sindex-delete: timeout while dropping %s:%s in SMD",
				ns_str, index_name_str);
		as_info_respond_error(db, AS_ERR_TIMEOUT, "timeout");
		return;
	}

	cf_free(fsk.found_key);

	as_info_respond_ok(db);
}

static void
cmd_sindex_exists(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// sindex-exists:ns=usermap;indexname=um_state

	char index_name_str[INAME_MAX_SZ];
	int index_name_len = sizeof(index_name_str);
	int rv = as_info_parameter_get(params, "indexname", index_name_str,
			&index_name_len);

	rv = as_info_required_param_is_ok(db, "indexname", index_name_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	char ns_str[AS_ID_NAMESPACE_SZ];
	int ns_len = sizeof(ns_str);
	as_namespace* ns = NULL;

	rv = as_info_param_get_namespace_ns(params, ns_str, &ns_len);

	if (! info_param_required_local_namespace_is_ok(db, ns_str, &ns, rv)) {
		return;
	}

	cf_dyn_buf_append_string(db, as_sindex_exists(ns, index_name_str) ?
			"true" : "false");
}

static void
cmd_sindex_list(const char* name, const char* params, cf_dyn_buf* db)
{
	char ns_str[128];
	int ns_len = sizeof(ns_str);
	info_param_result rv = as_info_param_get_namespace_ns(params, ns_str, &ns_len);
	as_namespace* ns = NULL;

	rv = info_param_optional_local_namespace_is_ok(db, ns_str, &ns, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	char b64_str[6];
	int b64_len = sizeof(b64_str);
	bool b64 = false;

	rv = as_info_parameter_get(params, "b64", b64_str, &b64_len);
	rv = as_info_optional_param_is_ok(db, "b64", b64_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK) {
		if (strcmp(b64_str, "true") == 0) {
			b64 = true;
		}
		else if (strcmp(b64_str, "false") == 0) {
			b64 = false;
		}
		else {
			cf_warning(AS_INFO, "b64 value invalid");
			as_info_respond_error(db, AS_ERR_PARAMETER, "bad b64");
			return;
		}
	}

	if (ns == NULL) {
		for (uint32_t ns_ix = 0; ns_ix < g_config.n_namespaces; ns_ix++) {
			as_sindex_list_str(g_config.namespaces[ns_ix], b64, db);
		}

		cf_dyn_buf_chomp_char(db, ';');
	}
	else {
		as_sindex_list_str(ns, b64, db);
		cf_dyn_buf_chomp_char(db, ';');
	}
}

static void
cmd_sindex_stat(const char* name, const char* params, cf_dyn_buf* db)
{
	// get indexname
	char index_name_str[INAME_MAX_SZ];
	int  index_len = sizeof(index_name_str);
	info_param_result rv = as_info_parameter_get(params, "indexname", index_name_str,
			&index_len);
	rv = as_info_required_param_is_ok(db, "indexname", index_name_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	char ns_str[AS_ID_NAMESPACE_SZ];
	int ns_len = sizeof(ns_str);
	as_namespace* ns = NULL;

	rv = as_info_param_get_namespace_ns(params, ns_str, &ns_len);

	if (! info_param_required_local_namespace_is_ok(db, ns_str, &ns, rv)) {
		return;
	}

	cf_info(AS_INFO, "sindex-stat - received request on index %s - namespace %s",
			index_name_str, ns_str);

	if (! as_sindex_stats_str(ns, index_name_str, db))  {
		cf_warning(AS_INFO, "index %s not found for ns %s",
				index_name_str, ns->name);
		as_info_respond_error(db, AS_ERR_SINDEX_NOT_FOUND, "no index");
	}
}

static void
cmd_smd_info(const char* name, const char* params, cf_dyn_buf* db)
{
	as_smd_get_info(db);
}

static void
cmd_smd_show(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format:
	// smd-show:module=sindex

	char module_str[8 + 1];
	int module_len = sizeof(module_str);
	info_param_result rv = as_info_parameter_get(params, "module", module_str,
			&module_len);

	if (! as_info_required_param_is_ok(db, "module", module_str, rv)) {
		return;
	}

	if (strcasecmp(module_str, "evict") == 0) {
		as_smd_get_all(AS_SMD_MODULE_EVICT, smd_show_cb, db);
	}
	else if(strcasecmp(module_str, "roster") == 0) {
		if (as_info_respond_enterprise_only(db)) {
			return;
		}

		as_smd_get_all(AS_SMD_MODULE_ROSTER, smd_show_cb, db);
	}
	else if(strcasecmp(module_str, "security") == 0) {
		if (as_info_respond_enterprise_only(db)) {
			return;
		}

		cf_warning(AS_INFO, "smd-show: security module forbidden");
		as_info_respond_error(db, AS_ERR_FORBIDDEN,
				"security module forbidden");
	}
	else if(strcasecmp(module_str, "sindex") == 0) {
		as_smd_get_all(AS_SMD_MODULE_SINDEX, smd_show_cb, db);
	}
	else if(strcasecmp(module_str, "truncate") == 0) {
		as_smd_get_all(AS_SMD_MODULE_TRUNCATE, smd_show_cb, db);
	}
	else if(strcasecmp(module_str, "UDF") == 0) {
		as_smd_get_all(AS_SMD_MODULE_UDF, smd_show_cb, db);
	}
	else if(strcasecmp(module_str, "XDR") == 0) {
		if (as_info_respond_enterprise_only(db)) {
			return;
		}

		as_smd_get_all(AS_SMD_MODULE_XDR, smd_show_cb, db);
	}
	else {
		cf_warning(AS_INFO, "smd-show: unknown 'module' %s", module_str);
		as_info_respond_error(db, AS_ERR_PARAMETER, "unknown 'module'");
	}

}

static void
cmd_statistics(const char* name, const char* params, cf_dyn_buf* db)
{
	uint64_t now_sec = cf_get_seconds();

	info_append_bool(db, "failed_best_practices", g_bad_practices.used_sz != 0);

	as_exchange_cluster_info(db);
	info_append_uint32(db, "cluster_min_compatibility_id", as_exchange_min_compatibility_id()); // not in ticker
	info_append_uint32(db, "cluster_max_compatibility_id", as_exchange_max_compatibility_id()); // not in ticker
	info_append_bool(db, "cluster_integrity", as_clustering_has_integrity()); // not in ticker
	info_append_bool(db, "cluster_is_member", ! as_clustering_is_orphan()); // not in ticker
	as_hb_info_duplicates_get(db); // not in ticker
	info_append_uint32(db, "cluster_clock_skew_stop_writes_sec", clock_skew_stop_writes_sec()); // not in ticker
	info_append_uint64(db, "cluster_clock_skew_ms", as_skew_monitor_skew());
	as_skew_monitor_info(db);

	info_append_uint64(db, "uptime", now_sec - g_start_sec); // not in ticker

	uint32_t user_pct = as_load_uint32(&g_user_cpu_pct);
	uint32_t kernel_pct = as_load_uint32(&g_kernel_cpu_pct);

	info_append_uint32(db, "system_total_cpu_pct", user_pct + kernel_pct);
	info_append_uint32(db, "system_user_cpu_pct", user_pct);
	info_append_uint32(db, "system_kernel_cpu_pct", kernel_pct);

	uint64_t free_mem_kbytes;
	uint32_t free_mem_pct;
	uint64_t thp_mem_kbytes;

	sys_mem_info(&free_mem_kbytes, &free_mem_pct, &thp_mem_kbytes);
	info_append_uint64(db, "system_free_mem_kbytes", free_mem_kbytes);
	info_append_int(db, "system_free_mem_pct", free_mem_pct);
	info_append_uint64(db, "system_thp_mem_kbytes", thp_mem_kbytes);

	info_append_uint32(db, "process_cpu_pct", g_process_cpu_pct);

	cf_thread_stats ts;

	cf_thread_get_stats(&ts);
	info_append_uint32(db, "threads_joinable", ts.n_joinable);
	info_append_uint32(db, "threads_detached", ts.n_detached);
	info_append_uint32(db, "threads_pool_total", ts.n_pool_total);
	info_append_uint32(db, "threads_pool_active", ts.n_pool_active);

	size_t allocated_kbytes;
	size_t active_kbytes;
	size_t mapped_kbytes;
	double efficiency_pct;
	uint32_t site_count;

	cf_alloc_heap_stats(&allocated_kbytes, &active_kbytes, &mapped_kbytes, &efficiency_pct,
			&site_count);
	info_append_uint64(db, "heap_allocated_kbytes", allocated_kbytes);
	info_append_uint64(db, "heap_active_kbytes", active_kbytes);
	info_append_uint64(db, "heap_mapped_kbytes", mapped_kbytes);
	info_append_int(db, "heap_efficiency_pct", (int)(efficiency_pct + 0.5));
	info_append_uint32(db, "heap_site_count", site_count);

	info_get_aggregated_namespace_stats(db);

	info_append_uint32(db, "info_queue", as_info_queue_get_size());
	info_append_uint32(db, "rw_in_progress", rw_request_hash_count());
	info_append_uint32(db, "proxy_in_progress", as_proxy_hash_count());
	info_append_uint32(db, "tree_gc_queue", as_index_tree_gc_queue_size());

	// Read closed before opened.
	uint64_t n_proto_fds_closed = as_load_uint64(&g_stats.proto_connections_closed);
	uint64_t n_hb_fds_closed = as_load_uint64(&g_stats.heartbeat_connections_closed);
	uint64_t n_fabric_fds_closed = as_load_uint64(&g_stats.fabric_connections_closed);
	// TODO - ARM TSO plugin - will need barrier.
	uint64_t n_proto_fds_opened = as_load_uint64(&g_stats.proto_connections_opened);
	uint64_t n_hb_fds_opened = as_load_uint64(&g_stats.heartbeat_connections_opened);
	uint64_t n_fabric_fds_opened = as_load_uint64(&g_stats.fabric_connections_opened);

	uint64_t n_proto_fds_open = n_proto_fds_opened - n_proto_fds_closed;
	uint64_t n_hb_fds_open = n_hb_fds_opened - n_hb_fds_closed;
	uint64_t n_fabric_fds_open = n_fabric_fds_opened - n_fabric_fds_closed;

	info_append_uint64(db, "client_connections", n_proto_fds_open);
	info_append_uint64(db, "client_connections_opened", n_proto_fds_opened);
	info_append_uint64(db, "client_connections_closed", n_proto_fds_closed);
	info_append_uint64(db, "heartbeat_connections", n_hb_fds_open);
	info_append_uint64(db, "heartbeat_connections_opened", n_hb_fds_opened);
	info_append_uint64(db, "heartbeat_connections_closed", n_hb_fds_closed);
	info_append_uint64(db, "fabric_connections", n_fabric_fds_open);
	info_append_uint64(db, "fabric_connections_opened", n_fabric_fds_opened);
	info_append_uint64(db, "fabric_connections_closed", n_fabric_fds_closed);

	info_append_uint64(db, "heartbeat_received_self", g_stats.heartbeat_received_self);
	info_append_uint64(db, "heartbeat_received_foreign", g_stats.heartbeat_received_foreign);

	info_append_uint64(db, "reaped_fds", g_stats.reaper_count); // not in ticker

	info_append_uint64(db, "info_complete", g_stats.info_complete); // not in ticker
	info_append_uint64(db, "info_timeout", g_stats.info_timeout); // not in ticker

	info_append_uint64(db, "demarshal_error", g_stats.n_demarshal_error);
	info_append_uint64(db, "early_tsvc_client_error", g_stats.n_tsvc_client_error);
	info_append_uint64(db, "early_tsvc_from_proxy_error", g_stats.n_tsvc_from_proxy_error);
	info_append_uint64(db, "early_tsvc_from_proxy_batch_sub_error", g_stats.n_tsvc_from_proxy_batch_sub_error);

	info_append_uint32(db, "long_queries_active", as_query_manager_get_active_job_count());

	info_append_uint64(db, "batch_index_initiate", g_stats.batch_index_initiate); // not in ticker

	cf_dyn_buf_append_string(db, "batch_index_queue=");
	as_batch_queues_info(db); // not in ticker
	cf_dyn_buf_append_char(db, ';');

	info_append_uint64(db, "batch_index_complete", g_stats.batch_index_complete);
	info_append_uint64(db, "batch_index_error", g_stats.batch_index_errors);
	info_append_uint64(db, "batch_index_timeout", g_stats.batch_index_timeout);
	info_append_uint64(db, "batch_index_delay", g_stats.batch_index_delay);

	// Everything below is not in ticker...

	info_append_uint32(db, "batch_index_unused_buffers", as_batch_unused_buffers());
	info_append_uint64(db, "batch_index_huge_buffers", g_stats.batch_index_huge_buffers);
	info_append_uint64(db, "batch_index_created_buffers", g_stats.batch_index_created_buffers);
	info_append_uint64(db, "batch_index_destroyed_buffers", g_stats.batch_index_destroyed_buffers);

	double batch_orig_sz = as_load_double(&g_stats.batch_comp_stat.avg_orig_sz);
	double batch_ratio = batch_orig_sz > 0.0 ? g_stats.batch_comp_stat.avg_comp_sz / batch_orig_sz : 1.0;

	info_append_format(db, "batch_index_proto_uncompressed_pct", "%.3f", g_stats.batch_comp_stat.uncomp_pct);
	info_append_format(db, "batch_index_proto_compression_ratio", "%.3f", batch_ratio);

	char paxos_principal[16 + 1];
	sprintf(paxos_principal, "%lX", as_exchange_principal());
	info_append_string(db, "paxos_principal", paxos_principal);

	info_append_uint64(db, "time_since_rebalance", now_sec - g_rebalance_sec);

	info_append_bool(db, "migrate_allowed", as_partition_balance_are_migrations_allowed());
	info_append_uint64(db, "migrate_partitions_remaining", as_partition_balance_remaining_migrations());

	info_append_uint64(db, "fabric_bulk_send_rate", g_stats.fabric_bulk_s_rate);
	info_append_uint64(db, "fabric_bulk_recv_rate", g_stats.fabric_bulk_r_rate);
	info_append_uint64(db, "fabric_ctrl_send_rate", g_stats.fabric_ctrl_s_rate);
	info_append_uint64(db, "fabric_ctrl_recv_rate", g_stats.fabric_ctrl_r_rate);
	info_append_uint64(db, "fabric_meta_send_rate", g_stats.fabric_meta_s_rate);
	info_append_uint64(db, "fabric_meta_recv_rate", g_stats.fabric_meta_r_rate);
	info_append_uint64(db, "fabric_rw_send_rate", g_stats.fabric_rw_s_rate);
	info_append_uint64(db, "fabric_rw_recv_rate", g_stats.fabric_rw_r_rate);

	cf_dyn_buf_chomp(db);
}

static void
cmd_status(const char* name, const char* params, cf_dyn_buf* db)
{
	as_info_respond_ok(db);
}

static void
cmd_thread_traces(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_thread_traces(name, db);
}

static void
cmd_tip(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: tip:host=<IPAddr>;port=<PortNum>[;tls=<Bool>]
	// where <IPAddr> is an IP address and <PortNum> is a valid TCP port number.

	cf_debug(AS_INFO, "tip command received: params %s", params);

	char host_str[DNS_NAME_MAX_SIZE];
	int  host_str_len = sizeof(host_str);

	info_param_result rv = as_info_parameter_get(params, "host", host_str,
			&host_str_len);

	if (! as_info_required_param_is_ok(db, "host", host_str, rv)) {
		return;
	}

	char port_str[50];
	int  port_str_len = sizeof(port_str);

	rv = as_info_parameter_get(params, "port", port_str, &port_str_len);

	if (! as_info_required_param_is_ok(db, "port", port_str, rv)) {
		return;
	}

	char tls_str[50];
	int  tls_str_len = sizeof(tls_str);

	rv = as_info_parameter_get(params, "tls", tls_str, &tls_str_len);
	rv = as_info_optional_param_is_ok(db, "tls", tls_str, rv);

	if (rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	if (rv == INFO_PARAM_OK_NOT_FOUND) {
		strcpy(tls_str, "false");
	}

	int port = 0;

	if (cf_str_atoi(port_str, &port) != 0) {
		cf_warning(AS_INFO, "tip command: port must be an integer in: %s",
				port_str);
		as_info_respond_error(db, AS_ERR_PARAMETER,
				"'port' must be an integer");
		return;
	}

	bool tls;
	if (strcmp(tls_str, "true") == 0) {
		tls = true;
	}
	else if (strcmp(tls_str, "false") == 0) {
		tls = false;
	}
	else {
		cf_warning(AS_INFO, "The '%s:' command argument 'tls' value must be one of {'true', 'false'}, not '%s'",
				name, tls_str);
		as_info_respond_error(db, AS_ERR_PARAMETER,
				"'tls' must be either 'true' or 'false'");
		return;
	}

	if (as_hb_mesh_tip(host_str, port, tls) == 0) {
		as_info_respond_ok(db);
	}
	else {
		as_info_respond_error(db, AS_ERR_UNKNOWN, "unknown");
	}
}

static void
cmd_tip_clear(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: tip-clear:{host-port-list=<hpl>}
	// "host-port-list" argument is optional]
	// where <hpl> is either "all" or else a comma-separated list of items
	// of the form: <HostIPv4Addr>:<PortNum> or [<HostIPv6Addr>]:<PortNum>

	cf_info(AS_INFO, "tip clear command received: params %s", params);

	char host_port_list[3000];
	int host_port_list_len = sizeof(host_port_list);
	bool success = true;
	uint32_t cleared = 0;
	uint32_t not_found = 0;
	info_param_result rv = as_info_parameter_get(params, "host-port-list",
			host_port_list, &host_port_list_len);

	if (! as_info_required_param_is_ok(db, "host-port-list", host_port_list,
			rv)) {
		return;
	}

	char* save_ptr = NULL;
	int port = -1;
	char* host_port = strtok_r(host_port_list, ",", &save_ptr);

	while (host_port != NULL) {
		char* host_port_delim = ":";

		if (*host_port == '[') {
			// Parse IPv6 address differently.
			host_port++;
			host_port_delim = "]";
		}

		char* host_port_save_ptr = NULL;
		char* host = strtok_r(host_port, host_port_delim, &host_port_save_ptr);

		if (host == NULL) {
			cf_warning(AS_INFO, "tip clear command: invalid host:port string: %s",
					host_port);
			success = false;
			break;
		}

		char* port_str = strtok_r(NULL, host_port_delim, &host_port_save_ptr);

		if (port_str != NULL && *port_str == ':') {
			// IPv6 case
			port_str++;
		}

		if (port_str == NULL || cf_str_atoi(port_str, &port) != 0) {
			cf_warning(AS_INFO, "tip clear command: port must be an integer in: %s",
					port_str);
			success = false;
			break;
		}

		if (as_hb_mesh_tip_clear(host, port) == -1) {
			success = false;
			not_found++;
			cf_warning(AS_INFO, "seed node %s:%d does not exist", host, port);
		}
		else {
			cleared++;
		}

		host_port = strtok_r(NULL, ",", &save_ptr);
	}

	if (success) {
		cf_info(AS_INFO, "tip clear command executed: cleared %u, params %s",
				cleared, params);
		as_info_respond_ok(db);
	}
	else {
		cf_info(AS_INFO, "tip clear command failed: cleared %u, params %s",
				cleared, params);
		as_info_respond_error(db, AS_ERR_UNKNOWN, "%u cleared, %u not found",
				cleared, not_found);
	}
}

static void
cmd_truncate(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: truncate:namespace=<ns-name>;set=<set-name>[;lut=<UTC-nanosec-string>]
	// where no lut value means use this server's current time.

	// Get the namespace name.

	char ns_name[AS_ID_NAMESPACE_SZ];
	int ns_name_len = (int)sizeof(ns_name);
	info_param_result ns_rv = as_info_param_get_namespace(params, ns_name,
			&ns_name_len);

	if (! as_info_required_param_is_ok(db, "namespace", ns_name, ns_rv)) {
		return;
	}

	// Get the set-name.

	char set_name[AS_SET_NAME_MAX_SIZE];
	int set_name_len = (int)sizeof(set_name);
	info_param_result set_rv = as_info_parameter_get(params, "set", set_name,
			&set_name_len);

	if (! as_info_required_param_is_ok(db, "set", set_name, set_rv)) {
		return;
	}

	// Get the threshold last-update-time, if there is one.

	char lut_str[24]; // allow decimal, hex or octal in C constant format
	int lut_str_len = (int)sizeof(lut_str);
	info_param_result lut_rv = as_info_parameter_get(params, "lut", lut_str,
			&lut_str_len);

	lut_rv = as_info_optional_param_is_ok(db, "lut", lut_str, lut_rv);

	if (lut_rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	// Issue the truncate command.

	as_truncate_cmd(ns_name, set_name, lut_rv == INFO_PARAM_OK ? lut_str : NULL,
			db);
}

static void
cmd_truncate_namespace(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: truncate-namespace:namespace=<ns-name>[;lut=<UTC-nanosec-string>]
	// where no lut value means use this server's current time.

	// Get the namespace name.

	char ns_name[AS_ID_NAMESPACE_SZ];
	int ns_name_len = (int)sizeof(ns_name);
	info_param_result ns_rv = as_info_param_get_namespace(params, ns_name,
			&ns_name_len);

	if (! as_info_required_param_is_ok(db, "namespace", ns_name, ns_rv)) {
		return;
	}

	// Check for a set-name, for safety. (Did user intend 'truncate'?)

	char set_name[1]; // just checking for existence
	int set_name_len = (int)sizeof(set_name);
	info_param_result set_rv = as_info_parameter_get(params, "set", set_name,
			&set_name_len);

	if (set_rv != INFO_PARAM_FAIL_NOT_FOUND) {
		cf_warning(AS_INFO, "truncate-namespace command: unexpected set name in command");
		as_info_respond_error(db, AS_ERR_PARAMETER, "unexpected set name");
		return;
	}

	// Get the threshold last-update-time, if there is one.

	char lut_str[24]; // allow decimal, hex or octal in C constant format
	int lut_str_len = (int)sizeof(lut_str);
	info_param_result lut_rv = as_info_parameter_get(params, "lut", lut_str,
			&lut_str_len);

	lut_rv = as_info_optional_param_is_ok(db, "lut", lut_str, lut_rv);

	if (lut_rv == INFO_PARAM_FAIL_REPLIED) {
		return;
	}

	// Issue the truncate command.

	as_truncate_cmd(ns_name, NULL, lut_rv == INFO_PARAM_OK ? lut_str : NULL,
			db);
}

static void
cmd_truncate_namespace_undo(const char* name, const char* params,
		cf_dyn_buf* db)
{
	// Command format: truncate-namespace-undo:namespace=<ns-name>

	// Get the namespace name.

	char ns_name[AS_ID_NAMESPACE_SZ];
	int ns_name_len = (int)sizeof(ns_name);
	info_param_result ns_rv = as_info_param_get_namespace(params, ns_name,
			&ns_name_len);

	if (! as_info_required_param_is_ok(db, "namespace", ns_name, ns_rv)) {
		return;
	}

	// Check for a set-name, for safety. (Did user intend 'truncate-undo'?)

	char set_name[1]; // just checking for existence
	int set_name_len = (int)sizeof(set_name);
	info_param_result set_rv = as_info_parameter_get(params, "set", set_name,
			&set_name_len);

	if (set_rv != INFO_PARAM_FAIL_NOT_FOUND) {
		cf_warning(AS_INFO, "truncate-namespace-undo command: unexpected set name in command");
		as_info_respond_error(db, AS_ERR_PARAMETER, "unexpected set name");
		return;
	}

	// Issue the truncate-undo command.

	as_truncate_undo_cmd(ns_name, NULL, db);
}

static void
cmd_truncate_undo(const char* name, const char* params, cf_dyn_buf* db)
{
	// Command format: truncate-undo:namespace=<ns-name>;set=<set-name>

	// Get the namespace name.

	char ns_name[AS_ID_NAMESPACE_SZ];
	int ns_name_len = (int)sizeof(ns_name);
	info_param_result ns_rv = as_info_param_get_namespace(params, ns_name,
			&ns_name_len);

	if (! as_info_required_param_is_ok(db, "namespace", ns_name, ns_rv)) {
		return;
	}

	// Get the set-name.

	char set_name[AS_SET_NAME_MAX_SIZE];
	int set_name_len = (int)sizeof(set_name);
	info_param_result set_rv = as_info_parameter_get(params, "set", set_name,
			&set_name_len);

	if (! as_info_required_param_is_ok(db, "set", set_name, set_rv)) {
		return;
	}

	// Issue the truncate-undo command.

	as_truncate_undo_cmd(ns_name, set_name, db);
}

static void
cmd_version(const char* name, const char* params, cf_dyn_buf* db)
{
	cf_dyn_buf_append_format(db, "%s build %s",
			aerospike_build_type, aerospike_build_id);
}


//==========================================================
// Local helpers - Info command helpers.
//

static const char*
perm_to_string(as_sec_perm perm)
{
	switch (perm) {
	case PERM_NONE:
		return "none";
	case PERM_READ:
		return "read";
	case PERM_QUERY:
		return "query";
	case PERM_WRITE:
		return "write";
	case PERM_DELETE:
		return "delete";
	case PERM_UDF_APPLY:
		return "udf-apply";
	case PERM_UDF_QUERY:
		return "udf-query";
	case PERM_OPS_QUERY:
		return "ops-query";
	case PERM_UDF_ADMIN:
		return "udf-admin";
	case PERM_SINDEX_ADMIN:
		return "sindex-admin";
	case PERM_QUERY_ADMIN:
		return "query-admin";
	case PERM_RECORD_INFO:
		return "record-info";
	case PERM_TRUNCATE:
		return "truncate";
	case PERM_EVICT_ADMIN:
		return "evict-admin";
	case PERM_SET_CONFIG:
		return "set-config";
	case PERM_LOGGING_CTRL:
		return "logging-ctrl";
	case PERM_SERVICE_CTRL:
		return "service-ctrl";
	case PERM_XDR_SET_FILTER:
		return "xdr-set-filter";
	case PERM_USER_ADMIN:
		return "urder-admin";
	default:
		cf_crash(AS_INFO, "unexpected perm %lu", perm);
	}

	return NULL;
}

static cf_ip_port
bind_to_port(cf_serv_cfg* cfg, cf_sock_owner owner)
{
	for (uint32_t i = 0; i < cfg->n_cfgs; ++i) {
		if (cfg->cfgs[i].owner == owner) {
			return cfg->cfgs[i].port;
		}
	}

	return 0;
}

static char*
access_to_string(cf_addr_list* addrs)
{
	cf_dyn_buf_define_size(db, 2500);

	for (uint32_t i = 0; i < addrs->n_addrs; ++i) {
		if (i > 0) {
			cf_dyn_buf_append_char(&db, ',');
		}

		cf_dyn_buf_append_string(&db, addrs->addrs[i]);
	}

	char* string = cf_dyn_buf_strdup(&db);

	cf_dyn_buf_free(&db);
	return string != NULL ? string : cf_strdup("null");
}

static void
info_get_aggregated_namespace_stats(cf_dyn_buf* db)
{
	uint64_t total_objects = 0;
	uint64_t total_tombstones = 0;

	for (uint32_t i = 0; i < g_config.n_namespaces; i++) {
		as_namespace* ns = g_config.namespaces[i];

		total_objects += ns->n_objects;
		total_tombstones += ns->n_tombstones;
	}

	info_append_uint64(db, "objects", total_objects);
	info_append_uint64(db, "tombstones", total_tombstones);
}

static void
info_get_namespace_info(as_namespace* ns, cf_dyn_buf* db)
{
	// Cluster size.

	// Using ns_ prefix to avoid confusion with global cluster_size.
	info_append_uint32(db, "ns_cluster_size", ns->cluster_size);

	info_append_uint32(db, "effective_replication_factor", ns->replication_factor);

	// Object counts.

	info_append_uint64(db, "objects", ns->n_objects);
	info_append_uint64(db, "tombstones", ns->n_tombstones);
	info_append_uint64(db, "xdr_tombstones", ns->n_xdr_tombstones);
	info_append_uint64(db, "xdr_bin_cemeteries", ns->n_xdr_bin_cemeteries);
	info_append_uint64(db, "mrt_provisionals", ns->n_mrt_provisionals);

	repl_stats mp;
	as_partition_get_replica_stats(ns, &mp);

	info_append_uint64(db, "master_objects", mp.n_master_objects);
	info_append_uint64(db, "master_tombstones", mp.n_master_tombstones);
	info_append_uint64(db, "prole_objects", mp.n_prole_objects);
	info_append_uint64(db, "prole_tombstones", mp.n_prole_tombstones);
	info_append_uint64(db, "non_replica_objects", mp.n_non_replica_objects);
	info_append_uint64(db, "non_replica_tombstones", mp.n_non_replica_tombstones);

	// Consistency info.

	info_append_uint64(db, "unreplicated_records", ns->n_unreplicated_records);
	info_append_uint32(db, "dead_partitions", ns->n_dead_partitions);
	info_append_uint32(db, "unavailable_partitions", ns->n_unavailable_partitions);
	info_append_uint32(db, "auto_revived_partitions", ns->n_auto_revived_partitions);
	info_append_bool(db, "clock_skew_stop_writes", ns->clock_skew_stop_writes);

	// Expiration & eviction (nsup) stats.

	info_append_bool(db, "stop_writes", ns->stop_writes);
	info_append_bool(db, "hwm_breached", ns->hwm_breached);

	info_append_uint64(db, "current_time", as_record_void_time_get());
	info_append_uint64(db, "non_expirable_objects", ns->non_expirable_objects);
	info_append_uint64(db, "expired_objects", ns->n_expired_objects);
	info_append_uint64(db, "evicted_objects", ns->n_evicted_objects);
	info_append_int(db, "evict_ttl", ns->evict_ttl);
	info_append_uint32(db, "evict_void_time", ns->evict_void_time);
	info_append_uint32(db, "smd_evict_void_time", ns->smd_evict_void_time);
	info_append_uint32(db, "nsup_cycle_duration", ns->nsup_cycle_duration);
	info_append_format(db, "nsup_cycle_deleted_pct", "%.2f", ns->nsup_cycle_deleted_pct);
	info_append_uint64(db, "nsup_xdr_key_busy", ns->n_nsup_xdr_key_busy);

	// Truncate stats.

	info_append_uint64(db, "truncate_lut", ns->truncate_lut);
	info_append_bool(db, "truncating", ns->truncating);

	// Sindex GC stats.

	info_append_uint64(db, "sindex_gc_cleaned", ns->n_sindex_gc_cleaned);

	// MRT monitor activity.

	info_append_uint32(db, "mrt_monitors_active", as_mrt_monitor_n_active(ns));

	// Persistent memory block keys' namespace ID (enterprise only).
	info_append_uint32(db, "xmem_id", ns->xmem_id);

	// Combined indexes stats.

	uint64_t ixs_memory_used = 0; // sum all RAM indexes

	// Primary index stats.

	uint64_t index_used = (ns->n_tombstones + ns->n_objects) * sizeof(as_index);

	info_append_uint64(db, "index_used_bytes", index_used);

	if (as_namespace_index_persisted(ns)) {
		info_append_uint64(db, "index_mounts_used_pct",
				index_used * 100 / ns->pi_mounts_budget);
	}
	else {
		ixs_memory_used += index_used;
	}

	if (ns->pi_xmem_type == CF_XMEM_TYPE_PMEM) {
		// If numa-pinned, not all configured mounts are used.
		if (as_config_is_numa_pinned()) {
			for (uint32_t i = 0; i < ns->n_pi_xmem_mounts; i++) {
				if (cf_mount_is_local(ns->pi_xmem_mounts[i])) {
					info_append_indexed_string(db, "index_local_mount", i, NULL,
							ns->pi_xmem_mounts[i]);
				}
			}
		}
	}
	else if (ns->pi_xmem_type == CF_XMEM_TYPE_FLASH) {
		uint64_t alloc_sz = as_load_uint64(&ns->arena->alloc_sz);

		info_append_uint64(db, "index_flash_alloc_bytes", alloc_sz);
		info_append_uint64(db, "index_flash_alloc_pct",
				alloc_sz * 100 / ns->pi_mounts_budget);

		add_index_device_stats(ns, db);
	}

	// Set index stats.

	uint64_t set_index_used = as_set_index_used_bytes(ns);

	info_append_uint64(db, "set_index_used_bytes", set_index_used);

	ixs_memory_used += set_index_used;

	// Secondary index stats.

	uint64_t sindex_used = as_sindex_used_bytes(ns);

	info_append_uint64(db, "sindex_used_bytes", sindex_used);

	if (as_namespace_sindex_persisted(ns)) {
		info_append_uint64(db, "sindex_mounts_used_pct",
				sindex_used * 100 / ns->si_mounts_budget);
	}
	else {
		ixs_memory_used += sindex_used;
	}

	if (ns->si_xmem_type == CF_XMEM_TYPE_PMEM) {
		// If numa-pinned, not all configured mounts are used.
		if (as_config_is_numa_pinned()) {
			for (uint32_t i = 0; i < ns->n_si_xmem_mounts; i++) {
				if (cf_mount_is_local(ns->si_xmem_mounts[i])) {
					info_append_indexed_string(db, "sindex_local_mount", i,
							NULL, ns->si_xmem_mounts[i]);
				}
			}
		}
	}
	else if (ns->si_xmem_type == CF_XMEM_TYPE_FLASH) {
		add_sindex_device_stats(ns, db);
	}

	// Combined indexes stats.

	uint64_t ixs_memory_budget = as_load_uint64(&ns->indexes_memory_budget);

	if (ixs_memory_budget != 0) {
		info_append_uint64(db, "indexes_memory_used_pct",
				ixs_memory_used * 100 / ixs_memory_budget);
	}

	// Storage stats.

	info_append_uint64(db, "data_total_bytes", ns->drives_size);

	uint32_t avail_pct = 0;
	uint64_t used_bytes = 0;

	as_storage_stats(ns, &avail_pct, &used_bytes);

	info_append_uint64(db, "data_used_bytes", used_bytes);
	info_append_uint64(db, "data_used_pct", used_bytes * 100 / ns->drives_size);
	info_append_uint32(db, "data_avail_pct", avail_pct);

	double orig_sz = as_load_double(&ns->comp_avg_orig_sz);
	double ratio = orig_sz > 0.0 ? ns->comp_avg_comp_sz / orig_sz : 1.0;

	info_append_format(db, "data_compression_ratio", "%.3f", ratio);

	if (ns->storage_type == AS_STORAGE_ENGINE_MEMORY) {
		add_data_stripe_stats(ns, db);
	}
	else if (ns->storage_type == AS_STORAGE_ENGINE_PMEM) {
		add_data_device_stats(ns, db);
	}
	else if (ns->storage_type == AS_STORAGE_ENGINE_SSD) {
		info_append_int(db, "cache_read_pct", (int)(ns->cache_read_pct + 0.5));

		add_data_device_stats(ns, db);
	}

	// Proto compression stats.

	double record_orig_sz = as_load_double(&ns->record_comp_stat.avg_orig_sz);
	double record_ratio = record_orig_sz > 0.0 ? ns->record_comp_stat.avg_comp_sz / record_orig_sz : 1.0;

	info_append_format(db, "record_proto_uncompressed_pct", "%.3f", ns->record_comp_stat.uncomp_pct);
	info_append_format(db, "record_proto_compression_ratio", "%.3f", record_ratio);

	double query_orig_sz = as_load_double(&ns->query_comp_stat.avg_orig_sz);
	double query_ratio = query_orig_sz > 0.0 ? ns->query_comp_stat.avg_comp_sz / query_orig_sz : 1.0;

	info_append_format(db, "query_proto_uncompressed_pct", "%.3f", ns->query_comp_stat.uncomp_pct);
	info_append_format(db, "query_proto_compression_ratio", "%.3f", query_ratio);

	// Partition balance state.

	as_exchange_info_lock();

	info_append_bool(db, "pending_quiesce", ns->pending_quiesce);
	info_append_bool(db, "effective_is_quiesced", ns->is_quiesced);
	info_append_uint64(db, "nodes_quiesced", ns->cluster_size - ns->active_size);

	info_append_bool(db, "effective_prefer_uniform_balance", ns->prefer_uniform_balance);
	info_append_uint32(db, "effective_active_rack", ns->cp ? ns->roster_active_rack : ns->active_rack);

	as_exchange_info_unlock();

	// Migration stats.

	info_append_uint64(db, "migrate_tx_partitions_imbalance", ns->migrate_tx_partitions_imbalance);

	info_append_uint64(db, "migrate_tx_instances", ns->migrate_tx_instance_count);
	info_append_uint64(db, "migrate_rx_instances", ns->migrate_rx_instance_count);

	info_append_uint64(db, "migrate_tx_partitions_active", ns->migrate_tx_partitions_active);
	info_append_uint64(db, "migrate_rx_partitions_active", ns->migrate_rx_partitions_active);

	info_append_uint64(db, "migrate_tx_partitions_initial", ns->migrate_tx_partitions_initial);
	info_append_uint64(db, "migrate_tx_partitions_remaining", ns->migrate_tx_partitions_remaining);
	info_append_uint64(db, "migrate_tx_partitions_lead_remaining", ns->migrate_tx_partitions_lead_remaining);

	info_append_uint64(db, "migrate_rx_partitions_initial", ns->migrate_rx_partitions_initial);
	info_append_uint64(db, "migrate_rx_partitions_remaining", ns->migrate_rx_partitions_remaining);

	info_append_uint64(db, "migrate_records_skipped", ns->migrate_records_skipped);
	info_append_uint64(db, "migrate_records_transmitted", ns->migrate_records_transmitted);
	info_append_uint64(db, "migrate_record_retransmits", ns->migrate_record_retransmits);
	info_append_uint64(db, "migrate_record_receives", ns->migrate_record_receives);
	info_append_uint64(db, "migrate_records_unreadable", ns->migrate_records_unreadable);

	info_append_uint64(db, "migrate_signals_active", ns->migrate_signals_active);
	info_append_uint64(db, "migrate_signals_remaining", ns->migrate_signals_remaining);

	info_append_uint64(db, "migrate_fresh_partitions", ns->migrate_fresh_partitions);

	info_append_uint64(db, "appeals_tx_active", ns->appeals_tx_active);
	info_append_uint64(db, "appeals_rx_active", ns->appeals_rx_active);

	info_append_uint64(db, "appeals_tx_remaining", ns->appeals_tx_remaining);

	info_append_uint64(db, "appeals_records_exonerated", ns->appeals_records_exonerated);

	// From-client transaction stats.

	info_append_uint64(db, "client_tsvc_error", ns->n_client_tsvc_error);
	info_append_uint64(db, "client_tsvc_timeout", ns->n_client_tsvc_timeout);

	info_append_uint64(db, "client_proxy_complete", ns->n_client_proxy_complete);
	info_append_uint64(db, "client_proxy_error", ns->n_client_proxy_error);
	info_append_uint64(db, "client_proxy_timeout", ns->n_client_proxy_timeout);

	info_append_uint64(db, "client_read_success", ns->n_client_read_success);
	info_append_uint64(db, "client_read_error", ns->n_client_read_error);
	info_append_uint64(db, "client_read_timeout", ns->n_client_read_timeout);
	info_append_uint64(db, "client_read_not_found", ns->n_client_read_not_found);
	info_append_uint64(db, "client_read_filtered_out", ns->n_client_read_filtered_out);

	info_append_uint64(db, "client_write_success", ns->n_client_write_success);
	info_append_uint64(db, "client_write_error", ns->n_client_write_error);
	info_append_uint64(db, "client_write_timeout", ns->n_client_write_timeout);
	info_append_uint64(db, "client_write_filtered_out", ns->n_client_write_filtered_out);

	// Subset of n_client_write_... above, respectively.
	info_append_uint64(db, "xdr_client_write_success", ns->n_xdr_client_write_success);
	info_append_uint64(db, "xdr_client_write_error", ns->n_xdr_client_write_error);
	info_append_uint64(db, "xdr_client_write_timeout", ns->n_xdr_client_write_timeout);

	info_append_uint64(db, "client_delete_success", ns->n_client_delete_success);
	info_append_uint64(db, "client_delete_error", ns->n_client_delete_error);
	info_append_uint64(db, "client_delete_timeout", ns->n_client_delete_timeout);
	info_append_uint64(db, "client_delete_not_found", ns->n_client_delete_not_found);
	info_append_uint64(db, "client_delete_filtered_out", ns->n_client_delete_filtered_out);

	// Subset of n_client_delete_... above, respectively.
	info_append_uint64(db, "xdr_client_delete_success", ns->n_xdr_client_delete_success);
	info_append_uint64(db, "xdr_client_delete_error", ns->n_xdr_client_delete_error);
	info_append_uint64(db, "xdr_client_delete_timeout", ns->n_xdr_client_delete_timeout);
	info_append_uint64(db, "xdr_client_delete_not_found", ns->n_xdr_client_delete_not_found);

	info_append_uint64(db, "client_udf_complete", ns->n_client_udf_complete);
	info_append_uint64(db, "client_udf_error", ns->n_client_udf_error);
	info_append_uint64(db, "client_udf_timeout", ns->n_client_udf_timeout);
	info_append_uint64(db, "client_udf_filtered_out", ns->n_client_udf_filtered_out);

	info_append_uint64(db, "client_lang_read_success", ns->n_client_lang_read_success);
	info_append_uint64(db, "client_lang_write_success", ns->n_client_lang_write_success);
	info_append_uint64(db, "client_lang_delete_success", ns->n_client_lang_delete_success);
	info_append_uint64(db, "client_lang_error", ns->n_client_lang_error);

	// From-proxy transaction stats.

	info_append_uint64(db, "from_proxy_tsvc_error", ns->n_from_proxy_tsvc_error);
	info_append_uint64(db, "from_proxy_tsvc_timeout", ns->n_from_proxy_tsvc_timeout);

	info_append_uint64(db, "from_proxy_read_success", ns->n_from_proxy_read_success);
	info_append_uint64(db, "from_proxy_read_error", ns->n_from_proxy_read_error);
	info_append_uint64(db, "from_proxy_read_timeout", ns->n_from_proxy_read_timeout);
	info_append_uint64(db, "from_proxy_read_not_found", ns->n_from_proxy_read_not_found);
	info_append_uint64(db, "from_proxy_read_filtered_out", ns->n_from_proxy_read_filtered_out);

	info_append_uint64(db, "from_proxy_write_success", ns->n_from_proxy_write_success);
	info_append_uint64(db, "from_proxy_write_error", ns->n_from_proxy_write_error);
	info_append_uint64(db, "from_proxy_write_timeout", ns->n_from_proxy_write_timeout);
	info_append_uint64(db, "from_proxy_write_filtered_out", ns->n_from_proxy_write_filtered_out);

	// Subset of n_from_proxy_write_... above, respectively.
	info_append_uint64(db, "xdr_from_proxy_write_success", ns->n_xdr_from_proxy_write_success);
	info_append_uint64(db, "xdr_from_proxy_write_error", ns->n_xdr_from_proxy_write_error);
	info_append_uint64(db, "xdr_from_proxy_write_timeout", ns->n_xdr_from_proxy_write_timeout);

	info_append_uint64(db, "from_proxy_delete_success", ns->n_from_proxy_delete_success);
	info_append_uint64(db, "from_proxy_delete_error", ns->n_from_proxy_delete_error);
	info_append_uint64(db, "from_proxy_delete_timeout", ns->n_from_proxy_delete_timeout);
	info_append_uint64(db, "from_proxy_delete_not_found", ns->n_from_proxy_delete_not_found);
	info_append_uint64(db, "from_proxy_delete_filtered_out", ns->n_from_proxy_delete_filtered_out);

	// Subset of n_from_proxy_delete_... above, respectively.
	info_append_uint64(db, "xdr_from_proxy_delete_success", ns->n_xdr_from_proxy_delete_success);
	info_append_uint64(db, "xdr_from_proxy_delete_error", ns->n_xdr_from_proxy_delete_error);
	info_append_uint64(db, "xdr_from_proxy_delete_timeout", ns->n_xdr_from_proxy_delete_timeout);
	info_append_uint64(db, "xdr_from_proxy_delete_not_found", ns->n_xdr_from_proxy_delete_not_found);

	info_append_uint64(db, "from_proxy_udf_complete", ns->n_from_proxy_udf_complete);
	info_append_uint64(db, "from_proxy_udf_error", ns->n_from_proxy_udf_error);
	info_append_uint64(db, "from_proxy_udf_timeout", ns->n_from_proxy_udf_timeout);
	info_append_uint64(db, "from_proxy_udf_filtered_out", ns->n_from_proxy_udf_filtered_out);

	info_append_uint64(db, "from_proxy_lang_read_success", ns->n_from_proxy_lang_read_success);
	info_append_uint64(db, "from_proxy_lang_write_success", ns->n_from_proxy_lang_write_success);
	info_append_uint64(db, "from_proxy_lang_delete_success", ns->n_from_proxy_lang_delete_success);
	info_append_uint64(db, "from_proxy_lang_error", ns->n_from_proxy_lang_error);

	// Batch sub-transaction stats.

	info_append_uint64(db, "batch_sub_tsvc_error", ns->n_batch_sub_tsvc_error);
	info_append_uint64(db, "batch_sub_tsvc_timeout", ns->n_batch_sub_tsvc_timeout);

	info_append_uint64(db, "batch_sub_proxy_complete", ns->n_batch_sub_proxy_complete);
	info_append_uint64(db, "batch_sub_proxy_error", ns->n_batch_sub_proxy_error);
	info_append_uint64(db, "batch_sub_proxy_timeout", ns->n_batch_sub_proxy_timeout);

	info_append_uint64(db, "batch_sub_read_success", ns->n_batch_sub_read_success);
	info_append_uint64(db, "batch_sub_read_error", ns->n_batch_sub_read_error);
	info_append_uint64(db, "batch_sub_read_timeout", ns->n_batch_sub_read_timeout);
	info_append_uint64(db, "batch_sub_read_not_found", ns->n_batch_sub_read_not_found);
	info_append_uint64(db, "batch_sub_read_filtered_out", ns->n_batch_sub_read_filtered_out);

	info_append_uint64(db, "batch_sub_write_success", ns->n_batch_sub_write_success);
	info_append_uint64(db, "batch_sub_write_error", ns->n_batch_sub_write_error);
	info_append_uint64(db, "batch_sub_write_timeout", ns->n_batch_sub_write_timeout);
	info_append_uint64(db, "batch_sub_write_filtered_out", ns->n_batch_sub_write_filtered_out);

	info_append_uint64(db, "batch_sub_delete_success", ns->n_batch_sub_delete_success);
	info_append_uint64(db, "batch_sub_delete_error", ns->n_batch_sub_delete_error);
	info_append_uint64(db, "batch_sub_delete_timeout", ns->n_batch_sub_delete_timeout);
	info_append_uint64(db, "batch_sub_delete_not_found", ns->n_batch_sub_delete_not_found);
	info_append_uint64(db, "batch_sub_delete_filtered_out", ns->n_batch_sub_delete_filtered_out);

	info_append_uint64(db, "batch_sub_udf_complete", ns->n_batch_sub_udf_complete);
	info_append_uint64(db, "batch_sub_udf_error", ns->n_batch_sub_udf_error);
	info_append_uint64(db, "batch_sub_udf_timeout", ns->n_batch_sub_udf_timeout);
	info_append_uint64(db, "batch_sub_udf_filtered_out", ns->n_batch_sub_udf_filtered_out);

	info_append_uint64(db, "batch_sub_lang_read_success", ns->n_batch_sub_lang_read_success);
	info_append_uint64(db, "batch_sub_lang_write_success", ns->n_batch_sub_lang_write_success);
	info_append_uint64(db, "batch_sub_lang_delete_success", ns->n_batch_sub_lang_delete_success);
	info_append_uint64(db, "batch_sub_lang_error", ns->n_batch_sub_lang_error);

	// From-proxy batch sub-transaction stats.

	info_append_uint64(db, "from_proxy_batch_sub_tsvc_error", ns->n_from_proxy_batch_sub_tsvc_error);
	info_append_uint64(db, "from_proxy_batch_sub_tsvc_timeout", ns->n_from_proxy_batch_sub_tsvc_timeout);

	info_append_uint64(db, "from_proxy_batch_sub_read_success", ns->n_from_proxy_batch_sub_read_success);
	info_append_uint64(db, "from_proxy_batch_sub_read_error", ns->n_from_proxy_batch_sub_read_error);
	info_append_uint64(db, "from_proxy_batch_sub_read_timeout", ns->n_from_proxy_batch_sub_read_timeout);
	info_append_uint64(db, "from_proxy_batch_sub_read_not_found", ns->n_from_proxy_batch_sub_read_not_found);
	info_append_uint64(db, "from_proxy_batch_sub_read_filtered_out", ns->n_from_proxy_batch_sub_read_filtered_out);

	info_append_uint64(db, "from_proxy_batch_sub_write_success", ns->n_from_proxy_batch_sub_write_success);
	info_append_uint64(db, "from_proxy_batch_sub_write_error", ns->n_from_proxy_batch_sub_write_error);
	info_append_uint64(db, "from_proxy_batch_sub_write_timeout", ns->n_from_proxy_batch_sub_write_timeout);
	info_append_uint64(db, "from_proxy_batch_sub_write_filtered_out", ns->n_from_proxy_batch_sub_write_filtered_out);

	info_append_uint64(db, "from_proxy_batch_sub_delete_success", ns->n_from_proxy_batch_sub_delete_success);
	info_append_uint64(db, "from_proxy_batch_sub_delete_error", ns->n_from_proxy_batch_sub_delete_error);
	info_append_uint64(db, "from_proxy_batch_sub_delete_timeout", ns->n_from_proxy_batch_sub_delete_timeout);
	info_append_uint64(db, "from_proxy_batch_sub_delete_not_found", ns->n_from_proxy_batch_sub_delete_not_found);
	info_append_uint64(db, "from_proxy_batch_sub_delete_filtered_out", ns->n_from_proxy_batch_sub_delete_filtered_out);

	info_append_uint64(db, "from_proxy_batch_sub_udf_complete", ns->n_from_proxy_batch_sub_udf_complete);
	info_append_uint64(db, "from_proxy_batch_sub_udf_error", ns->n_from_proxy_batch_sub_udf_error);
	info_append_uint64(db, "from_proxy_batch_sub_udf_timeout", ns->n_from_proxy_batch_sub_udf_timeout);
	info_append_uint64(db, "from_proxy_batch_sub_udf_filtered_out", ns->n_from_proxy_batch_sub_udf_filtered_out);

	info_append_uint64(db, "from_proxy_batch_sub_lang_read_success", ns->n_from_proxy_batch_sub_lang_read_success);
	info_append_uint64(db, "from_proxy_batch_sub_lang_write_success", ns->n_from_proxy_batch_sub_lang_write_success);
	info_append_uint64(db, "from_proxy_batch_sub_lang_delete_success", ns->n_from_proxy_batch_sub_lang_delete_success);
	info_append_uint64(db, "from_proxy_batch_sub_lang_error", ns->n_from_proxy_batch_sub_lang_error);

	// Internal-UDF sub-transaction stats.

	info_append_uint64(db, "udf_sub_tsvc_error", ns->n_udf_sub_tsvc_error);
	info_append_uint64(db, "udf_sub_tsvc_timeout", ns->n_udf_sub_tsvc_timeout);

	info_append_uint64(db, "udf_sub_udf_complete", ns->n_udf_sub_udf_complete);
	info_append_uint64(db, "udf_sub_udf_error", ns->n_udf_sub_udf_error);
	info_append_uint64(db, "udf_sub_udf_timeout", ns->n_udf_sub_udf_timeout);
	info_append_uint64(db, "udf_sub_udf_filtered_out", ns->n_udf_sub_udf_filtered_out);

	info_append_uint64(db, "udf_sub_lang_read_success", ns->n_udf_sub_lang_read_success);
	info_append_uint64(db, "udf_sub_lang_write_success", ns->n_udf_sub_lang_write_success);
	info_append_uint64(db, "udf_sub_lang_delete_success", ns->n_udf_sub_lang_delete_success);
	info_append_uint64(db, "udf_sub_lang_error", ns->n_udf_sub_lang_error);

	// Internal-ops sub-transaction stats.

	info_append_uint64(db, "ops_sub_tsvc_error", ns->n_ops_sub_tsvc_error);
	info_append_uint64(db, "ops_sub_tsvc_timeout", ns->n_ops_sub_tsvc_timeout);

	info_append_uint64(db, "ops_sub_write_success", ns->n_ops_sub_write_success);
	info_append_uint64(db, "ops_sub_write_error", ns->n_ops_sub_write_error);
	info_append_uint64(db, "ops_sub_write_timeout", ns->n_ops_sub_write_timeout);
	info_append_uint64(db, "ops_sub_write_filtered_out", ns->n_ops_sub_write_filtered_out);

	// Duplicate resolution stats.

	info_append_uint64(db, "dup_res_ask", ns->n_dup_res_ask);

	info_append_uint64(db, "dup_res_respond_read", ns->n_dup_res_respond_read);
	info_append_uint64(db, "dup_res_respond_no_read", ns->n_dup_res_respond_no_read);

	// Transaction retransmit stats - 'all' means both client & proxy origins.

	info_append_uint64(db, "retransmit_all_read_dup_res", ns->n_retransmit_all_read_dup_res);
	info_append_uint64(db, "retransmit_all_write_dup_res", ns->n_retransmit_all_write_dup_res);
	info_append_uint64(db, "retransmit_all_delete_dup_res", ns->n_retransmit_all_delete_dup_res);
	info_append_uint64(db, "retransmit_all_udf_dup_res", ns->n_retransmit_all_udf_dup_res);
	info_append_uint64(db, "retransmit_all_batch_sub_read_dup_res", ns->n_retransmit_all_batch_sub_read_dup_res);
	info_append_uint64(db, "retransmit_all_batch_sub_write_dup_res", ns->n_retransmit_all_batch_sub_write_dup_res);
	info_append_uint64(db, "retransmit_all_batch_sub_delete_dup_res", ns->n_retransmit_all_batch_sub_delete_dup_res);
	info_append_uint64(db, "retransmit_all_batch_sub_udf_dup_res", ns->n_retransmit_all_batch_sub_udf_dup_res);
	info_append_uint64(db, "retransmit_udf_sub_dup_res", ns->n_retransmit_udf_sub_dup_res);
	info_append_uint64(db, "retransmit_ops_sub_dup_res", ns->n_retransmit_ops_sub_dup_res);

	info_append_uint64(db, "retransmit_all_read_repl_ping", ns->n_retransmit_all_read_repl_ping);
	info_append_uint64(db, "retransmit_all_batch_sub_read_repl_ping", ns->n_retransmit_all_batch_sub_read_repl_ping);

	info_append_uint64(db, "retransmit_all_write_repl_write", ns->n_retransmit_all_write_repl_write);
	info_append_uint64(db, "retransmit_all_delete_repl_write", ns->n_retransmit_all_delete_repl_write);
	info_append_uint64(db, "retransmit_all_udf_repl_write", ns->n_retransmit_all_udf_repl_write);
	info_append_uint64(db, "retransmit_all_batch_sub_write_repl_write", ns->n_retransmit_all_batch_sub_write_repl_write);
	info_append_uint64(db, "retransmit_all_batch_sub_delete_repl_write", ns->n_retransmit_all_batch_sub_delete_repl_write);
	info_append_uint64(db, "retransmit_all_batch_sub_udf_repl_write", ns->n_retransmit_all_batch_sub_udf_repl_write);
	info_append_uint64(db, "retransmit_udf_sub_repl_write", ns->n_retransmit_udf_sub_repl_write);
	info_append_uint64(db, "retransmit_ops_sub_repl_write", ns->n_retransmit_ops_sub_repl_write);

	// Primary index query (formerly scan) stats.

	info_append_uint64(db, "pi_query_short_basic_complete", ns->n_pi_query_short_basic_complete);
	info_append_uint64(db, "pi_query_short_basic_error", ns->n_pi_query_short_basic_error);
	info_append_uint64(db, "pi_query_short_basic_timeout", ns->n_pi_query_short_basic_timeout);

	info_append_uint64(db, "pi_query_long_basic_complete", ns->n_pi_query_long_basic_complete);
	info_append_uint64(db, "pi_query_long_basic_error", ns->n_pi_query_long_basic_error);
	info_append_uint64(db, "pi_query_long_basic_abort", ns->n_pi_query_long_basic_abort);

	info_append_uint64(db, "pi_query_aggr_complete", ns->n_pi_query_aggr_complete);
	info_append_uint64(db, "pi_query_aggr_error", ns->n_pi_query_aggr_error);
	info_append_uint64(db, "pi_query_aggr_abort", ns->n_pi_query_aggr_abort);

	info_append_uint64(db, "pi_query_udf_bg_complete", ns->n_pi_query_udf_bg_complete);
	info_append_uint64(db, "pi_query_udf_bg_error", ns->n_pi_query_udf_bg_error);
	info_append_uint64(db, "pi_query_udf_bg_abort", ns->n_pi_query_udf_bg_abort);

	info_append_uint64(db, "pi_query_ops_bg_complete", ns->n_pi_query_ops_bg_complete);
	info_append_uint64(db, "pi_query_ops_bg_error", ns->n_pi_query_ops_bg_error);
	info_append_uint64(db, "pi_query_ops_bg_abort", ns->n_pi_query_ops_bg_abort);

	// Secondary index query stats.

	info_append_uint64(db, "si_query_short_basic_complete", ns->n_si_query_short_basic_complete);
	info_append_uint64(db, "si_query_short_basic_error", ns->n_si_query_short_basic_error);
	info_append_uint64(db, "si_query_short_basic_timeout", ns->n_si_query_short_basic_timeout);

	info_append_uint64(db, "si_query_long_basic_complete", ns->n_si_query_long_basic_complete);
	info_append_uint64(db, "si_query_long_basic_error", ns->n_si_query_long_basic_error);
	info_append_uint64(db, "si_query_long_basic_abort", ns->n_si_query_long_basic_abort);

	info_append_uint64(db, "si_query_aggr_complete", ns->n_si_query_aggr_complete);
	info_append_uint64(db, "si_query_aggr_error", ns->n_si_query_aggr_error);
	info_append_uint64(db, "si_query_aggr_abort", ns->n_si_query_aggr_abort);

	info_append_uint64(db, "si_query_udf_bg_complete", ns->n_si_query_udf_bg_complete);
	info_append_uint64(db, "si_query_udf_bg_error", ns->n_si_query_udf_bg_error);
	info_append_uint64(db, "si_query_udf_bg_abort", ns->n_si_query_udf_bg_abort);

	info_append_uint64(db, "si_query_ops_bg_complete", ns->n_si_query_ops_bg_complete);
	info_append_uint64(db, "si_query_ops_bg_error", ns->n_si_query_ops_bg_error);
	info_append_uint64(db, "si_query_ops_bg_abort", ns->n_si_query_ops_bg_abort);

	// Geospatial query stats:
	info_append_uint64(db, "geo_region_query_reqs", ns->geo_region_query_count);
	info_append_uint64(db, "geo_region_query_cells", ns->geo_region_query_cells);
	info_append_uint64(db, "geo_region_query_points", ns->geo_region_query_points);
	info_append_uint64(db, "geo_region_query_falsepos", ns->geo_region_query_falsepos);

	// Read-touch stats.

	info_append_uint64(db, "read_touch_tsvc_error", ns->n_read_touch_tsvc_error);
	info_append_uint64(db, "read_touch_tsvc_timeout", ns->n_read_touch_tsvc_timeout);

	info_append_uint64(db, "read_touch_success", ns->n_read_touch_success);
	info_append_uint64(db, "read_touch_error", ns->n_read_touch_error);
	info_append_uint64(db, "read_touch_timeout", ns->n_read_touch_timeout);
	info_append_uint64(db, "read_touch_skip", ns->n_read_touch_skip);

	// Re-replication stats - relevant only for enterprise edition.

	info_append_uint64(db, "re_repl_tsvc_error", ns->n_re_repl_tsvc_error);
	info_append_uint64(db, "re_repl_tsvc_timeout", ns->n_re_repl_tsvc_timeout);

	info_append_uint64(db, "re_repl_success", ns->n_re_repl_success);
	info_append_uint64(db, "re_repl_error", ns->n_re_repl_error);
	info_append_uint64(db, "re_repl_timeout", ns->n_re_repl_timeout);

	// MRT verify read stats - relevant only for enterprise edition.

	info_append_uint64(db, "mrt_verify_read_success", ns->n_mrt_verify_read_success);
	info_append_uint64(db, "mrt_verify_read_error", ns->n_mrt_verify_read_error);
	info_append_uint64(db, "mrt_verify_read_timeout", ns->n_mrt_verify_read_timeout);

	// MRT roll forward/back stats - relevant only for enterprise edition.

	info_append_uint64(db, "mrt_roll_forward_success", ns->n_mrt_roll_forward_success);
	info_append_uint64(db, "mrt_roll_forward_error", ns->n_mrt_roll_forward_error);
	info_append_uint64(db, "mrt_roll_forward_timeout", ns->n_mrt_roll_forward_timeout);

	// Subset of n_mrt_roll_forward... above, respectively
	info_append_uint64(db, "mrt_monitor_roll_forward_success", ns->n_mrt_monitor_roll_forward_success);
	info_append_uint64(db, "mrt_monitor_roll_forward_error", ns->n_mrt_monitor_roll_forward_error);
	info_append_uint64(db, "mrt_monitor_roll_forward_timeout", ns->n_mrt_monitor_roll_forward_timeout);

	info_append_uint64(db, "mrt_roll_back_success", ns->n_mrt_roll_back_success);
	info_append_uint64(db, "mrt_roll_back_error", ns->n_mrt_roll_back_error);
	info_append_uint64(db, "mrt_roll_back_timeout", ns->n_mrt_roll_back_timeout);

	// Subset of n_mrt_roll_back... above, respectively
	info_append_uint64(db, "mrt_monitor_roll_back_success", ns->n_mrt_monitor_roll_back_success);
	info_append_uint64(db, "mrt_monitor_roll_back_error", ns->n_mrt_monitor_roll_back_error);
	info_append_uint64(db, "mrt_monitor_roll_back_timeout", ns->n_mrt_monitor_roll_back_timeout);

	// Special errors that deserve their own counters:

	info_append_uint64(db, "fail_xdr_forbidden", ns->n_fail_xdr_forbidden);
	info_append_uint64(db, "fail_key_busy", ns->n_fail_key_busy);
	info_append_uint64(db, "fail_xdr_key_busy", ns->n_fail_xdr_key_busy);
	info_append_uint64(db, "fail_generation", ns->n_fail_generation);
	info_append_uint64(db, "fail_record_too_big", ns->n_fail_record_too_big);
	info_append_uint64(db, "fail_client_lost_conflict", ns->n_fail_client_lost_conflict);
	info_append_uint64(db, "fail_xdr_lost_conflict", ns->n_fail_xdr_lost_conflict);
	info_append_uint64(db, "fail_mrt_blocked", ns->n_fail_mrt_blocked);
	info_append_uint64(db, "fail_mrt_version_mismatch", ns->n_fail_mrt_version_mismatch);

	// Special non-error counters:

	info_append_uint64(db, "deleted_last_bin", ns->n_deleted_last_bin);
	info_append_uint64(db, "mrt_monitor_roll_tombstone_creates", ns->n_mrt_monitor_roll_tombstone_creates);
}

static void
namespace_rack_info(as_namespace* ns, cf_dyn_buf* db, uint32_t* rack_ids,
		uint32_t n_nodes, cf_node node_seq[], const char* tag)
{
	if (n_nodes == 0) {
		return;
	}

	rack_node rack_nodes[n_nodes];

	for (uint32_t n = 0; n < n_nodes; n++) {
		rack_nodes[n].rack_id = rack_ids[n];
		rack_nodes[n].node = node_seq[n];
	}

	qsort(rack_nodes, n_nodes, sizeof(rack_node), compare_rack_nodes);

	uint32_t cur_id = rack_nodes[0].rack_id;

	cf_dyn_buf_append_string(db, tag);
	cf_dyn_buf_append_uint32(db, cur_id);
	cf_dyn_buf_append_char(db, '=');
	cf_dyn_buf_append_uint64_x(db, rack_nodes[0].node);

	for (uint32_t n = 1; n < n_nodes; n++) {
		if (rack_nodes[n].rack_id == cur_id) {
			cf_dyn_buf_append_char(db, ',');
			cf_dyn_buf_append_uint64_x(db, rack_nodes[n].node);
			continue;
		}

		cur_id = rack_nodes[n].rack_id;

		cf_dyn_buf_append_char(db, ':');
		cf_dyn_buf_append_string(db, tag);
		cf_dyn_buf_append_uint32(db, cur_id);
		cf_dyn_buf_append_char(db, '=');
		cf_dyn_buf_append_uint64_x(db, rack_nodes[n].node);
	}
}

static void
namespace_roster_info(as_namespace* ns, cf_dyn_buf* db)
{
	as_exchange_info_lock();

	cf_dyn_buf_append_string(db, "roster=");

	if (ns->roster_count == 0) {
		cf_dyn_buf_append_string(db, "null");
	}
	else {
		if (ns->roster_active_rack != 0) {
			cf_dyn_buf_append_format(db, "%c%u|", ROSTER_ACTIVE_RACK_TOK,
					ns->roster_active_rack);
		}

		for (uint32_t n = 0; n < ns->roster_count; n++) {
			cf_dyn_buf_append_uint64_x(db, ns->roster[n]);

			if (ns->roster_rack_ids[n] != 0) {
				cf_dyn_buf_append_char(db, ROSTER_ID_PAIR_SEPARATOR);
				cf_dyn_buf_append_uint32(db, ns->roster_rack_ids[n]);
			}

			cf_dyn_buf_append_char(db, ',');
		}

		cf_dyn_buf_chomp(db);
	}

	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "pending_roster=");

	if (ns->smd_roster_count == 0) {
		cf_dyn_buf_append_string(db, "null");
	}
	else {
		if (ns->smd_roster_active_rack != 0) {
			cf_dyn_buf_append_format(db, "%c%u|", ROSTER_ACTIVE_RACK_TOK,
					ns->smd_roster_active_rack);
		}

		for (uint32_t n = 0; n < ns->smd_roster_count; n++) {
			cf_dyn_buf_append_uint64_x(db, ns->smd_roster[n]);

			if (ns->smd_roster_rack_ids[n] != 0) {
				cf_dyn_buf_append_char(db, ROSTER_ID_PAIR_SEPARATOR);
				cf_dyn_buf_append_uint32(db, ns->smd_roster_rack_ids[n]);
			}

			cf_dyn_buf_append_char(db, ',');
		}

		cf_dyn_buf_chomp(db);
	}

	cf_dyn_buf_append_char(db, ':');

	cf_dyn_buf_append_string(db, "observed_nodes=");

	if (ns->observed_cluster_size == 0) {
		cf_dyn_buf_append_string(db, "null");
	}
	else {
		if (ns->active_rack != 0) {
			cf_dyn_buf_append_format(db, "%c%u|", ROSTER_ACTIVE_RACK_TOK,
					ns->active_rack);
		}

		for (uint32_t n = 0; n < ns->observed_cluster_size; n++) {
			cf_dyn_buf_append_uint64_x(db, ns->observed_succession[n]);

			if (ns->rack_ids[n] != 0) {
				cf_dyn_buf_append_char(db, ROSTER_ID_PAIR_SEPARATOR);
				cf_dyn_buf_append_uint32(db, ns->rack_ids[n]);
			}

			cf_dyn_buf_append_char(db, ',');
		}

		cf_dyn_buf_chomp(db);
	}

	as_exchange_info_unlock();
}

static void
add_index_device_stats(as_namespace* ns, cf_dyn_buf* db)
{
	for (uint32_t i = 0; i < ns->n_pi_xmem_mounts; i++) {
		info_append_indexed_int(db, "index-type.mount", i, "age",
				oldest_nvme_age(ns->pi_xmem_mounts[i]));
	}
}

static void
add_sindex_device_stats(as_namespace* ns, cf_dyn_buf* db)
{
	for (uint32_t i = 0; i < ns->n_si_xmem_mounts; i++) {
		info_append_indexed_int(db, "sindex-type.mount", i, "age",
				oldest_nvme_age(ns->si_xmem_mounts[i]));
	}
}

static int32_t
oldest_nvme_age(const char* path)
{
	cf_storage_device_info* info = cf_storage_get_device_info(path);

	if (info == NULL) {
		return -1;
	}

	int32_t oldest = -1;

	for (int32_t i = 0; i < info->n_phys; ++i) {
		if (info->phys[i].nvme_age > oldest) {
			oldest = info->phys[i].nvme_age;
		}
	}

	return oldest;
}

static void
add_data_stripe_stats(as_namespace* ns, cf_dyn_buf* db)
{
	const char* tag = "storage-engine.stripe";

	for (uint32_t i = 0; i < ns->n_storage_stripes; i++) {
		storage_device_stats stats;
		as_storage_device_stats(ns, i, &stats);

		info_append_indexed_uint64(db, tag, i, "used_bytes", stats.used_sz);
		info_append_indexed_uint32(db, tag, i, "free_wblocks", stats.n_free_wblocks);

		info_append_indexed_uint64(db, tag, i, "writes", stats.n_writes);

		info_append_indexed_uint32(db, tag, i, "defrag_q", stats.defrag_q_sz);
		info_append_indexed_uint64(db, tag, i, "defrag_reads", stats.n_defrag_reads);
		info_append_indexed_uint64(db, tag, i, "defrag_writes", stats.n_defrag_writes);

		if (ns->n_storage_shadows != 0) {
			info_append_indexed_uint32(db, tag, i, "backing_write_q", stats.shadow_write_q_sz);
			info_append_indexed_uint64(db, tag, i, "partial_writes", stats.n_partial_writes);
			info_append_indexed_uint64(db, tag, i, "defrag_partial_writes", stats.n_defrag_partial_writes);

			// Can't tell if this is local or remote - just try it.
			info_append_indexed_int(db, tag, i, "age",
					oldest_nvme_age(ns->storage_shadows[i]));
		}
	}
}

static void
add_data_device_stats(as_namespace* ns, cf_dyn_buf* db)
{
	uint32_t n = as_namespace_device_count(ns);
	const char* tag = ns->n_storage_devices != 0 ?
			"storage-engine.device" : "storage-engine.file";

	for (uint32_t i = 0; i < n; i++) {
		storage_device_stats stats;
		as_storage_device_stats(ns, i, &stats);

		info_append_indexed_uint64(db, tag, i, "used_bytes", stats.used_sz);
		info_append_indexed_uint32(db, tag, i, "free_wblocks", stats.n_free_wblocks);

		info_append_indexed_uint64(db, tag, i, "read_errors", stats.n_read_errors);

		info_append_indexed_uint32(db, tag, i, "write_q", stats.write_q_sz);
		info_append_indexed_uint64(db, tag, i, "writes", stats.n_writes);
		info_append_indexed_uint64(db, tag, i, "partial_writes", stats.n_partial_writes);

		info_append_indexed_uint32(db, tag, i, "defrag_q", stats.defrag_q_sz);
		info_append_indexed_uint64(db, tag, i, "defrag_reads", stats.n_defrag_reads);
		info_append_indexed_uint64(db, tag, i, "defrag_writes", stats.n_defrag_writes);
		info_append_indexed_uint64(db, tag, i, "defrag_partial_writes", stats.n_defrag_partial_writes);

		if (ns->n_storage_shadows != 0) {
			info_append_indexed_uint32(db, tag, i, "shadow_write_q", stats.shadow_write_q_sz);
		}

		info_append_indexed_int(db, tag, i, "age",
				oldest_nvme_age(ns->storage_devices[i]));

		// Note - no shadow ages since they're likely remote.
	}
}

static void
find_sindex_key(const cf_vector* items, void* udata)
{
	find_sindex_key_udata* fsk = (find_sindex_key_udata*)udata;
	uint32_t ns_name_len = strlen(fsk->ns_name);

	fsk->found_key = NULL;
	fsk->n_name_matches = 0;
	fsk->n_indexes = 0;
	fsk->has_smd_key = false;

	for (uint32_t i = 0; i < cf_vector_size(items); i++) {
		as_smd_item* item = cf_vector_get_ptr(items, i);

		if (item->value == NULL) {
			continue; // ignore tombstones
		}

		const char* smd_ns_name_end = strchr(item->key, '|');

		if (smd_ns_name_end == NULL) {
			cf_warning(AS_INFO, "unexpected sindex key format '%s'", item->key);
			continue;
		}

		uint32_t smd_ns_name_len = smd_ns_name_end - item->key;

		if (smd_ns_name_len != ns_name_len ||
				memcmp(item->key, fsk->ns_name, ns_name_len) != 0) {
			continue;
		}

		fsk->n_indexes++;

		if (fsk->smd_key != NULL && strcmp(fsk->smd_key, item->key) == 0) {
			fsk->has_smd_key = true;
			fsk->smd_key = NULL; // can only be one
		}

		if (strcmp(fsk->index_name, item->value) != 0) {
			continue;
		}

		fsk->n_name_matches++;

		if (fsk->n_name_matches == 1) {
			fsk->found_key = strdup(item->key);
		}
		else {
			cf_free(fsk->found_key); // only return when unique
			fsk->found_key = NULL;
		}
	}
}

static void
smd_show_cb(const cf_vector* items, void* udata)
{
	cf_dyn_buf* db = (cf_dyn_buf*)udata;
	uint32_t n_items = 0;

	for (uint32_t i = 0; i < cf_vector_size(items); i++) {
		as_smd_item* item = cf_vector_get_ptr(items, i);

		if (item->value == NULL) {
			continue; // ignore tombstones
		}

		n_items++;

		cf_dyn_buf_append_string(db, item->key);
		cf_dyn_buf_append_char(db, '='); // for now, not escaping
		cf_dyn_buf_append_string(db, item->value);
		cf_dyn_buf_append_char(db, ';');
	}

	if (n_items != 0) {
		cf_dyn_buf_chomp_char(db, ';');
	}
	else {
		cf_dyn_buf_append_string(db, "<empty>");
	}
}

// TODO - these utilities are like the info_append_...() set but with commas -
// move them elsewhere? (Like dynbuf?)

static inline void
db_append_uint32(cf_dyn_buf* db, const char* name, uint32_t value)
{
	cf_dyn_buf_append_string(db, name);
	cf_dyn_buf_append_char(db, '=');
	cf_dyn_buf_append_uint32(db, value);
	cf_dyn_buf_append_char(db, ',');
}

static inline void
db_append_uint64(cf_dyn_buf* db, const char* name, uint64_t value)
{
	cf_dyn_buf_append_string(db, name);
	cf_dyn_buf_append_char(db, '=');
	cf_dyn_buf_append_uint64(db, value);
	cf_dyn_buf_append_char(db, ',');
}

static inline void
db_append_uint64_x(cf_dyn_buf* db, const char* name, uint64_t value)
{
	cf_dyn_buf_append_string(db, name);
	cf_dyn_buf_append_char(db, '=');
	cf_dyn_buf_append_uint64_x(db, value);
	cf_dyn_buf_append_char(db, ',');
}

static inline void
db_append_string_safe(cf_dyn_buf* db, const char* name, const char* value)
{
	cf_dyn_buf_append_string(db, name);
	cf_dyn_buf_append_char(db, '=');
	cf_dyn_buf_append_string(db, value ? value : "null");
	cf_dyn_buf_append_char(db, ',');
}

static inline void
db_append_format(cf_dyn_buf* db, const char* name, const char* form, ...)
{
	va_list va;
	va_start(va, form);

	cf_dyn_buf_append_string(db, name);
	cf_dyn_buf_append_char(db, '=');
	cf_dyn_buf_append_format_va(db, form, va);
	cf_dyn_buf_append_char(db, ',');

	va_end(va);
}

static inline void
db_append_hex(cf_dyn_buf* db, const char* name, const uint8_t* buf, uint32_t sz,
		char delimiter)
{
	cf_dyn_buf_append_string(db, name);
	cf_dyn_buf_append_char(db, '=');

	for (uint32_t i = 0; i < sz; i++) {
		cf_dyn_buf_append_format(db, "%02x", buf[i]);
	}

	cf_dyn_buf_append_char(db, delimiter);
}

static void
debug_record(const char* params, cf_dyn_buf* db, bool all_data)
{
	// Get the namespace name.

	char ns_name[AS_ID_NAMESPACE_SZ];
	int ns_name_len = (int)sizeof(ns_name);
	info_param_result ns_rv = as_info_param_get_namespace(params, ns_name,
			&ns_name_len);
	as_namespace* ns;

	if (! info_param_required_local_namespace_is_ok(db, ns_name, &ns, ns_rv)) {
		return;
	}

	// Get the digest.

	char keyd_str[(CF_DIGEST_KEY_SZ * 2) + 1];
	int keyd_str_len = (int)sizeof(keyd_str);
	info_param_result keyd_rv = as_info_parameter_get(params, "keyd", keyd_str,
			&keyd_str_len);

	if (! as_info_required_param_is_ok(db, "keyd", keyd_str, keyd_rv)) {
		return;
	}

	if (keyd_str_len != CF_DIGEST_KEY_SZ * 2) {
		cf_warning(AS_INFO, "debug-record: missing or invalid keyd");
		as_info_respond_error(db, AS_ERR_PARAMETER, "keyd");
		return;
	}

	char hex_byte[2 + 1] = { 0 };
	char* at = keyd_str;

	cf_digest keyd;

	for (uint32_t i = 0; i < CF_DIGEST_KEY_SZ; i++) {
		hex_byte[0] = *at++;
		hex_byte[1] = *at++;

		uint64_t val;

		if (cf_strtoul_x64(hex_byte, &val) != 0) {
			cf_warning(AS_INFO, "debug-record: bad keyd %s", keyd_str);
			as_info_respond_error(db, AS_ERR_PARAMETER, "bad keyd");
			return;
		}

		keyd.digest[i] = (uint8_t)val;
	}

	// Get the mode, if applicable.

	bool as_pickle = false;
	bool as_raw = false;
	bool leave_encrypted = false;

	if (all_data) {
		char mode[AS_ID_NAMESPACE_SZ];
		int mode_len = (int)sizeof(mode);
		info_param_result mode_rv = as_info_parameter_get(params, "mode", mode,
				&mode_len);

		mode_rv = as_info_optional_param_is_ok(db, "mode", mode, mode_rv);

		if (mode_rv == INFO_PARAM_FAIL_REPLIED) {
			return;
		}

		if (mode_rv == INFO_PARAM_OK) {
			if (strcmp(mode, "pickle") == 0) {
				as_pickle = true;
			}
			else if (strcmp(mode, "raw") == 0) {
				as_raw = true;
			}
			else if (strcmp(mode, "raw-encrypted") == 0) {
				as_raw = true;
				leave_encrypted = true;
			}
			else {
				cf_warning(AS_INFO, "debug-record: unknown mode");
				as_info_respond_error(db, AS_ERR_PARAMETER, "unknown mode");
				return;
			}
		}
	}

	// Get the record.

	as_partition_reservation rsv;
	uint32_t pid = as_partition_getid(&keyd);

	as_partition_reserve(ns, pid, &rsv);

	as_index_ref r_ref;

	if (as_record_get(rsv.tree, &keyd, &r_ref) != 0) {
		as_partition_release(&rsv);
		cf_warning(AS_INFO, "debug-record: %pD not found", &keyd);
		as_info_respond_error(db, AS_ERR_NOT_FOUND, "record not found");
		return;
	}

	info_append_uint32(db, "pid", pid);

	cf_mutex_lock(&rsv.p->lock); // hack using lock outside of partition code
	info_append_int(db, "repl-ix", find_self_in_replicas(rsv.p));
	cf_mutex_unlock(&rsv.p->lock);

	as_index* r = r_ref.r;

	debug_dump_index(db, ns, r, false);

	as_storage_rd rd;

	as_storage_record_open(ns, r, &rd);

	if (as_pickle) {
		debug_dump_pickle(db, &rd, false);
	}
	else if (as_raw) {
		debug_dump_raw(db, &rd, leave_encrypted, false);
	}
	else {
		debug_dump_parsed(db, &rd, all_data, false);
	}

	as_storage_record_close(&rd);

	if (r->orig_h != 0) {
		as_record* orig_r = cf_arenax_resolve(ns->arena, r->orig_h);

		debug_dump_index(db, ns, orig_r, true);

		if (orig_r->generation != 0) {
			as_storage_rd orig_rd;

			as_storage_record_open(ns, orig_r, &orig_rd);

			if (as_pickle) {
				debug_dump_pickle(db, &orig_rd, true);
			}
			else if (as_raw) {
				debug_dump_raw(db, &orig_rd, leave_encrypted, true);
			}
			else {
				debug_dump_parsed(db, &orig_rd, all_data, true);
			}

			as_storage_record_close(&orig_rd);
		}
	}

	as_record_done(&r_ref, ns);
	as_partition_release(&rsv);

	cf_dyn_buf_chomp(db);
}

static void
debug_dump_index(cf_dyn_buf* db, as_namespace* ns, as_record* r, bool is_orig)
{
	cf_dyn_buf_append_string(db, is_orig ? "orig-index=" : "index=");

	db_append_uint32(db, "rc", r->rc);
	db_append_uint32(db, "tree-id", r->tree_id);
	db_append_uint32(db, "color", r->color);
	db_append_uint32(db, "is-orig", r->is_orig);
	db_append_format(db, "keyd", "%D", &r->keyd); // not %pD (as used for logs)
	db_append_uint64_x(db, "left", r->left_h);
	db_append_uint64_x(db, "right", r->right_h);
	db_append_uint32(db, "set-id", r->set_id_bits);
	db_append_string_safe(db, "set-name",
			as_namespace_get_set_name(ns, r->set_id_bits));
	db_append_uint32(db, "in-sindex", r->in_sindex);
	db_append_uint32(db, "xdr-bin-cemetery", r->xdr_bin_cemetery);
	db_append_uint32(db, "xdr-write", r->xdr_write);
	db_append_uint32(db, "xdr-tombstone", r->xdr_tombstone);
	db_append_uint32(db, "xdr-nsup-tombstone", r->xdr_nsup_tombstone);
	db_append_uint32(db, "void-time", r->void_time);
	db_append_uint64(db, "lut", r->last_update_time);
	db_append_uint64(db, "generation", plain_generation(r->generation, ns));

	if (ns->cp) { // hack instead of split regime() wrapper
		db_append_uint64(db, "regime", r->generation >> 10); // GEN_BITS
	}

	db_append_uint64(db, "rblock-id", r->rblock_id);
	db_append_uint64(db, "n-rblocks", r->n_rblocks);
	db_append_uint64(db, "file-id", r->file_id);

	uint32_t n_devices = ns->n_storage_stripes != 0 ?
			ns->n_storage_stripes : as_namespace_device_count(ns);

	if (n_devices != 0 && r->file_id < n_devices) {
		const char* tag = ns->n_storage_stripes != 0 ?
				"stripe-name" : (ns->n_storage_devices != 0 ?
						"device-name" : "file-name");

		db_append_string_safe(db, tag, ns->storage_devices[r->file_id]);
	}

	db_append_uint64(db, "key-stored", r->key_stored);
	db_append_uint32(db, "repl-state", r->repl_state);
	db_append_uint32(db, "tombstone", r->tombstone);
	db_append_uint32(db, "cenotaph", r->cenotaph);

	if (is_orig) {
		info_append_uint64_x(db, "mrt-id", r->mrt_id);
	}
	else {
		info_append_uint64_x(db, "orig-h", (uint64_t)r->orig_h);
	}
}

static void
debug_dump_pickle(cf_dyn_buf* db, as_storage_rd* rd, bool is_orig)
{
	if (as_storage_record_load_pickle(rd)) {
		info_append_uint32(db, is_orig ? "orig-pickle-size" : "pickle-size",
				rd->pickle_sz);
		db_append_hex(db, is_orig ? "orig-pickle" : "pickle", rd->pickle,
				rd->pickle_sz, ';');
	}
}

static void
debug_dump_raw(cf_dyn_buf* db, as_storage_rd* rd, bool leave_encrypted,
		bool is_orig)
{
	if (as_storage_record_load_raw(rd, leave_encrypted)) {
		const uint8_t* raw = (const uint8_t*)rd->flat;
		uint32_t sz = (uint32_t)(rd->flat_end - raw);

		info_append_uint32(db, is_orig ? "orig-raw-size" : "raw-size", sz);
		db_append_hex(db, is_orig ? "orig-raw-bytes" : "raw-bytes", raw, sz,
				';');
	}
}

static void
debug_dump_parsed(cf_dyn_buf* db, as_storage_rd* rd, bool all_data,
		bool is_orig)
{
	// Note - for now, not parsing stored fields redundant with index.

	as_record* r = rd->r;

	if (r->key_stored == 1 && as_storage_rd_load_key(rd)) {
		// Start of key.
		cf_dyn_buf_append_string(db, is_orig ? "orig-key=" : "key=");

		uint8_t type = *rd->key;
		const uint8_t* key = rd->key + 1;
		uint32_t sz = rd->key_size - 1;

		db_append_uint32(db, "flat-size", rd->key_size);
		db_append_uint32(db, "type", type);

		if (all_data) {
			switch (type) {
			case AS_PARTICLE_TYPE_INTEGER:
				if (sz == sizeof(uint64_t)) {
					// Flat integer keys are in big-endian order.
					db_append_uint64(db, "value",
							cf_swap_from_be64(*(uint64_t*)key));
				}
				break;
			case AS_PARTICLE_TYPE_STRING:
			case AS_PARTICLE_TYPE_BLOB:
				// Apparently we can get CDTs here? And floats?!
			default:
				db_append_hex(db, "value", key, sz, ',');
				break;
			}
		}

		cf_dyn_buf_chomp(db);
		cf_dyn_buf_append_char(db, ';');
		// End of key.
	}

	as_bin stack_bins[RECORD_MAX_BINS];

	if (as_storage_rd_load_bins(rd, stack_bins) < 0) {
		cf_warning(AS_INFO, "debug-record: failed to load bins");
		return;
	}

	info_append_uint32(db, is_orig ? "orig-n-bins" : "n-bins", rd->n_bins);

	// Start of bins.
	if (rd->n_bins != 0) {
		cf_dyn_buf_append_string(db, is_orig ? "orig-bins=" : "bins=");
	}

	for (uint16_t i = 0; i < rd->n_bins; i++) {
		as_bin* b = &rd->bins[i];

		db_append_string_safe(db, "bin-name", b->name);
		db_append_uint32(db, "xdr-write", b->xdr_write);
		db_append_uint64(db, "lut", b->lut);
		db_append_uint32(db, "src-id", b->src_id);

		db_append_uint32(db, "state", b->state);

		switch (b->state) {
		case AS_BIN_STATE_UNUSED:
		case AS_BIN_STATE_TOMBSTONE:
			break;
		case AS_BIN_STATE_INUSE_INTEGER:
			db_append_uint32(db, "type", AS_PARTICLE_TYPE_INTEGER);
			if (all_data) {
				db_append_uint64(db, "value", (uint64_t)b->particle);
			}
			break;
		case AS_BIN_STATE_INUSE_FLOAT:
			db_append_uint32(db, "type", AS_PARTICLE_TYPE_FLOAT);
			if (all_data) {
				db_append_format(db, "value", "%f", *(double*)&b->particle);
			}
			break;
		case AS_BIN_STATE_INUSE_BOOL:
			db_append_uint32(db, "type", AS_PARTICLE_TYPE_BOOL);
			if (all_data) {
				db_append_uint32(db, "value", (uint8_t)(uint64_t)b->particle);
			}
			break;
		case AS_BIN_STATE_INUSE_OTHER:
			db_append_uint32(db, "type", b->particle->type);
			{
				uint32_t sz = *(uint32_t*)b->particle->data; // hack

				db_append_uint32(db, "data-size", sz);

				if (all_data) {
					db_append_hex(db, "value",
							b->particle->data + sizeof(uint32_t), sz, ',');
				}
			}
			break;
		default:
			cf_warning(AS_INFO, "debug-record: bad bin state %u", b->state);
			break;
		}

		cf_dyn_buf_chomp(db);
		cf_dyn_buf_append_char(db, ':'); // delimit multiple bins
	}
	// End of bins.

	cf_dyn_buf_chomp(db);
	cf_dyn_buf_append_char(db, ';');
}
